1+ import type { RedisClient } from 'bun'
2+ import { generateId } from './utils'
3+ import { createLogger } from './logger'
4+
5+ export interface LockOptions {
6+ /**
7+ * Lock duration in milliseconds (defaults to 30000 ms / 30 seconds)
8+ */
9+ duration ?: number
10+
11+ /**
12+ * Whether to automatically extend the lock as it approaches expiration (defaults to true)
13+ */
14+ autoExtend ?: boolean
15+
16+ /**
17+ * How often to extend the lock (defaults to 2/3 of the duration)
18+ */
19+ extendInterval ?: number
20+
21+ /**
22+ * Number of retries to acquire the lock if it's already locked (defaults to 0)
23+ */
24+ retries ?: number
25+
26+ /**
27+ * Delay between retries in milliseconds (defaults to 100ms)
28+ */
29+ retryDelay ?: number
30+ }
31+
32+ /**
33+ * Distributed lock implementation using Redis
34+ */
35+ export class DistributedLock {
36+ private redisClient : RedisClient
37+ private prefix : string
38+ private readonly logger = createLogger ( 'lock' )
39+
40+ constructor ( redisClient : RedisClient , prefix = 'lock' ) {
41+ this . redisClient = redisClient
42+ this . prefix = prefix
43+ }
44+
45+ /**
46+ * Acquire a lock
47+ * @param resource The resource to lock
48+ * @param options Lock options
49+ * @returns Lock token if successfully acquired, null otherwise
50+ */
51+ async acquire ( resource : string , options : LockOptions = { } ) : Promise < string | null > {
52+ const lockKey = this . getLockKey ( resource )
53+ const token = generateId ( )
54+ const duration = options . duration || 30000 // Default 30 seconds
55+ const retries = options . retries || 0
56+ const retryDelay = options . retryDelay || 100
57+
58+ // Try to acquire the lock
59+ for ( let attempt = 0 ; attempt <= retries ; attempt ++ ) {
60+ // Use SET NX (only set if key doesn't exist) with expiration
61+ const result = await this . redisClient . send ( 'SET' , [
62+ lockKey ,
63+ token ,
64+ 'NX' , // Only set if key doesn't exist
65+ 'PX' , // Set expiry in milliseconds
66+ duration . toString ( ) ,
67+ ] )
68+
69+ if ( result === 'OK' ) {
70+ this . logger . debug ( `Acquired lock ${ resource } ${ token } )
71+
72+ // Set up auto-extension if enabled
73+ if ( options . autoExtend !== false ) {
74+ const extendInterval = options . extendInterval || Math . floor ( duration * 2 / 3 )
75+ this . setupAutoExtend ( resource , token , duration , extendInterval )
76+ }
77+
78+ return token
79+ }
80+
81+ // If not successful and we have retries left
82+ if ( attempt < retries ) {
83+ await new Promise ( resolve => setTimeout ( resolve , retryDelay ) )
84+ }
85+ }
86+
87+ this . logger . debug ( `Failed to acquire lock ${ resource } ${ retries + 1 } )
88+ return null
89+ }
90+
91+ /**
92+ * Release a lock
93+ * @param resource The resource to unlock
94+ * @param token The lock token for validation
95+ * @returns True if successfully released, false otherwise
96+ */
97+ async release ( resource : string , token : string ) : Promise < boolean > {
98+ const lockKey = this . getLockKey ( resource )
99+
100+ // We need to implement the Lua script logic with normal Redis commands
101+ // since we can't use eval directly
102+
103+ // First, check if we own the lock
104+ const currentToken = await this . redisClient . get ( lockKey )
105+
106+ if ( currentToken === token ) {
107+ // We own the lock, so delete it
108+ await this . redisClient . del ( lockKey )
109+ this . logger . debug ( `Released lock ${ resource } ${ token } )
110+ return true
111+ } else {
112+ this . logger . debug ( `Failed to release lock ${ resource } ${ token } )
113+ return false
114+ }
115+ }
116+
117+ /**
118+ * Check if a lock exists without acquiring it
119+ * @param resource The resource to check
120+ * @returns True if locked, false otherwise
121+ */
122+ async isLocked ( resource : string ) : Promise < boolean > {
123+ const lockKey = this . getLockKey ( resource )
124+ const result = await this . redisClient . exists ( lockKey )
125+ // Different Redis clients may return different types
126+ return result ? true : false
127+ }
128+
129+ /**
130+ * Extend a lock's duration
131+ * @param resource The resource to extend
132+ * @param token The lock token for validation
133+ * @param duration New duration in milliseconds
134+ * @returns True if successfully extended, false otherwise
135+ */
136+ async extend ( resource : string , token : string , duration : number ) : Promise < boolean > {
137+ const lockKey = this . getLockKey ( resource )
138+
139+ // Implement Lua script logic with standard Redis commands
140+ const currentToken = await this . redisClient . get ( lockKey )
141+
142+ if ( currentToken === token ) {
143+ // We own the lock, so extend it
144+ const result = await this . redisClient . send ( 'PEXPIRE' , [ lockKey , duration . toString ( ) ] )
145+
146+ if ( result === 1 || result === true ) {
147+ this . logger . debug ( `Extended lock ${ resource } ${ token } ${ duration } )
148+ return true
149+ }
150+ }
151+
152+ this . logger . debug ( `Failed to extend lock ${ resource } ${ token } )
153+ return false
154+ }
155+
156+ /**
157+ * Set up automatic lock extension
158+ */
159+ private setupAutoExtend ( resource : string , token : string , duration : number , interval : number ) : void {
160+ const autoExtendId = setInterval ( async ( ) => {
161+ try {
162+ const extended = await this . extend ( resource , token , duration )
163+
164+ if ( ! extended ) {
165+ // Lock no longer exists or we don't own it anymore
166+ clearInterval ( autoExtendId )
167+ this . logger . debug ( `Stopped auto-extension for lock ${ resource } )
168+ }
169+ } catch ( error ) {
170+ this . logger . error ( `Error extending lock ${ resource } ${ ( error as Error ) . message } )
171+ clearInterval ( autoExtendId )
172+ }
173+ } , interval )
174+
175+ // Ensure we clean up the interval when Node exits
176+ const resourceKey = `${ resource } ${ token }
177+ if ( typeof process !== 'undefined' ) {
178+ const extendTimers = DistributedLock . autoExtendTimers
179+ extendTimers . set ( resourceKey , autoExtendId )
180+
181+ // Cleanup on first timer
182+ if ( extendTimers . size === 1 ) {
183+ process . once ( 'exit' , ( ) => {
184+ for ( const timer of extendTimers . values ( ) ) {
185+ clearInterval ( timer )
186+ }
187+ extendTimers . clear ( )
188+ } )
189+ }
190+ }
191+ }
192+
193+ /**
194+ * Get the lock key with prefix
195+ */
196+ private getLockKey ( resource : string ) : string {
197+ return `${ this . prefix } ${ resource }
198+ }
199+
200+ /**
201+ * Store interval timers for cleanup
202+ */
203+ private static autoExtendTimers = new Map < string , NodeJS . Timeout > ( )
204+
205+ /**
206+ * Execute a function with a lock
207+ * @param resource The resource to lock
208+ * @param fn The function to execute while holding the lock
209+ * @param options Lock options
210+ * @returns The result of the function
211+ * @throws Error if the lock cannot be acquired
212+ */
213+ async withLock < T > ( resource : string , fn : ( ) => Promise < T > , options : LockOptions = { } ) : Promise < T > {
214+ const token = await this . acquire ( resource , options )
215+
216+ if ( ! token ) {
217+ throw new Error ( `Failed to acquire lock for resource ${ resource } )
218+ }
219+
220+ try {
221+ return await fn ( )
222+ } finally {
223+ await this . release ( resource , token )
224+ }
225+ }
226+ }
0 commit comments