Live video on the AT Protocol
stream.place
1package statedb
2
3import (
4 "crypto/sha256"
5 "encoding/binary"
6 "fmt"
7 "sync"
8)
9
10func (state *StatefulDB) GetNamedLock(name string) (func(), error) {
11 // temp while we debug postgres
12 return state.getNamedLockSQLite(name)
13 // switch state.Type {
14 // case DBTypeSQLite:
15 // return state.getNamedLockSQLite(name)
16 // case DBTypePostgres:
17 // return state.getNamedLockPostgres(name)
18 // }
19 // panic("unsupported database type")
20}
21
22func (state *StatefulDB) getNamedLockPostgres(name string) (func(), error) {
23 // Convert string to sha256 hash and use decimal value for advisory lock
24 h := sha256.Sum256([]byte(name))
25 nameInt := int64(binary.BigEndian.Uint64(h[:8]))
26
27 err := state.DB.Exec("SELECT pg_advisory_lock($1)", nameInt).Error
28 if err != nil {
29 return nil, err
30 }
31 return func() {
32 err := state.DB.Exec("SELECT pg_advisory_unlock($1)", nameInt).Error
33 if err != nil {
34 // unfortunate, but the risk is that we're holding on to the lock forever,
35 // so it's responsible to crash in this case
36 panic(fmt.Errorf("error unlocking named lock: %w", err))
37 }
38 }, nil
39}
40
41func (state *StatefulDB) getNamedLockSQLite(name string) (func(), error) {
42 lock := state.locks.GetLock(name)
43 lock.Lock()
44 return func() {
45 lock.Unlock()
46 }, nil
47}
48
49// Local mutex implementation for sqlite
50type NamedLocks struct {
51 mu sync.Mutex
52 locks map[string]*sync.Mutex
53}
54
55// NewNamedLocks creates a new NamedLocks instance
56func NewNamedLocks() *NamedLocks {
57 return &NamedLocks{
58 locks: make(map[string]*sync.Mutex),
59 }
60}
61
62// GetLock returns the mutex for the given name, creating it if it doesn't exist
63func (n *NamedLocks) GetLock(name string) *sync.Mutex {
64 n.mu.Lock()
65 defer n.mu.Unlock()
66
67 lock, exists := n.locks[name]
68 if !exists {
69 lock = &sync.Mutex{}
70 n.locks[name] = lock
71 }
72 return lock
73}