1package storage
2
3import (
4 "context"
5 "database/sql"
6 "errors"
7 "fmt"
8 "io"
9 "net/url"
10 "os"
11 "path/filepath"
12 "strings"
13 "sync"
14 "sync/atomic"
15 "syscall"
16 "time"
17
18 "tug/internal/storage/internal/attach"
19)
20
21const (
22 busyTimeoutMillis = 5000
23 initializationLockSuffix = ".init.lock"
24)
25
26type Store struct {
27 sql *sql.DB
28 path string
29 now func() time.Time
30 newID func() (string, error)
31 closed atomic.Bool
32 closeOnce sync.Once
33 closeErr error
34 attachMu sync.Mutex
35 attached []attachment
36}
37
38func Open(ctx context.Context, databasePath string) (*Store, error) {
39 if strings.TrimSpace(databasePath) == "" {
40 return nil, invalid("open storage", fmt.Errorf("database path is empty"))
41 }
42 absPath, err := filepath.Abs(databasePath)
43 if err != nil {
44 return nil, invalid("open storage", fmt.Errorf("absolute database path: %w", err))
45 }
46 initializationLock, err := acquireInitializationLock(ctx, absPath)
47 if err != nil {
48 return nil, err
49 }
50 defer initializationLock.Close()
51 if err := createDatabaseFile(absPath); err != nil {
52 return nil, err
53 }
54
55 sqlDB, err := openSQLite(ctx, absPath)
56 if err != nil {
57 return nil, err
58 }
59 migrations, err := loadMigrations(embeddedMigrations, "migrations")
60 if err != nil {
61 sqlDB.Close()
62 return nil, fmt.Errorf("load embedded migrations: %w", err)
63 }
64 if err := applyMigrations(ctx, sqlDB, migrations, time.Now()); err != nil {
65 sqlDB.Close()
66 return nil, err
67 }
68
69 return &Store{
70 sql: sqlDB,
71 path: absPath,
72 now: time.Now,
73 newID: newUUID,
74 }, nil
75}
76
77func (s *Store) Attach(ctx context.Context, attacher attach.Attacher) error {
78 const operation = "attach storage"
79 if attacher == nil {
80 return invalid(operation, fmt.Errorf("attacher is nil"))
81 }
82 if s == nil || s.sql == nil {
83 return fmt.Errorf("%s: %w", operation, ErrClosed)
84 }
85 s.attachMu.Lock()
86 defer s.attachMu.Unlock()
87 if err := s.checkOpen(operation); err != nil {
88 return err
89 }
90 initializationLock, err := acquireInitializationLock(ctx, s.path)
91 if err != nil {
92 return fmt.Errorf("%s: %w", operation, err)
93 }
94 defer initializationLock.Close()
95 attached, err := runAttacher(ctx, s.path, attacher)
96 if err != nil {
97 return fmt.Errorf("%s: %w", operation, err)
98 }
99 if err := s.checkOpen(operation); err != nil {
100 attached.Close()
101 return err
102 }
103 s.attached = append(s.attached, attached)
104 return nil
105}
106
107type attachment struct {
108 closer io.Closer
109 pools []*sql.DB
110}
111
112func (a attachment) Close() error {
113 errs := []error{a.closer.Close()}
114 for _, pool := range a.pools {
115 errs = append(errs, pool.Close())
116 }
117 return errors.Join(errs...)
118}
119
120func runAttacher(ctx context.Context, absPath string, attacher attach.Attacher) (attachment, error) {
121 var mu sync.Mutex
122 active := true
123 var opened []*sql.DB
124 pool := attach.NewPool(func(ctx context.Context) (*sql.DB, error) {
125 mu.Lock()
126 defer mu.Unlock()
127 if !active {
128 return nil, fmt.Errorf("open attached sqlite: %w", ErrClosed)
129 }
130 db, err := openSQLite(ctx, absPath)
131 if err != nil {
132 return nil, err
133 }
134 opened = append(opened, db)
135 return db, nil
136 })
137 closer, err := attacher.Attach(ctx, pool)
138 mu.Lock()
139 active = false
140 mu.Unlock()
141 if err == nil && closer == nil {
142 err = errors.New("attacher returned a nil closer")
143 }
144 if err != nil {
145 if closer != nil {
146 closer.Close()
147 }
148 for _, db := range opened {
149 db.Close()
150 }
151 return attachment{}, err
152 }
153 return attachment{closer: closer, pools: opened}, nil
154}
155
156func openSQLite(ctx context.Context, absPath string) (*sql.DB, error) {
157 sqlDB, err := sql.Open("sqlite", sqliteURL(absPath))
158 if err != nil {
159 return nil, classify("open sqlite", err)
160 }
161 if err := sqlDB.PingContext(ctx); err != nil {
162 sqlDB.Close()
163 return nil, classify("ping sqlite", err)
164 }
165 if err := verifyPragmas(ctx, sqlDB); err != nil {
166 sqlDB.Close()
167 return nil, err
168 }
169 return sqlDB, nil
170}
171
172func sqliteURL(absPath string) string {
173 databaseURL := &url.URL{Scheme: "file", Path: filepath.ToSlash(absPath)}
174 query := url.Values{}
175 query.Add("_pragma", "foreign_keys(1)")
176 query.Add("_pragma", "journal_mode(WAL)")
177 query.Add("_pragma", fmt.Sprintf("busy_timeout(%d)", busyTimeoutMillis))
178 query.Set("_txlock", "immediate")
179 databaseURL.RawQuery = query.Encode()
180 return databaseURL.String()
181}
182
183func createDatabaseFile(absPath string) error {
184 file, err := os.OpenFile(absPath, os.O_CREATE|os.O_EXCL|os.O_RDWR, 0o600)
185 if errors.Is(err, os.ErrExist) {
186 return nil
187 }
188 if err != nil {
189 return classify("create database file", err)
190 }
191 return classify("create database file", file.Close())
192}
193
194type initializationLock struct {
195 file *os.File
196}
197
198func acquireInitializationLock(ctx context.Context, databasePath string) (*initializationLock, error) {
199 file, err := os.OpenFile(databasePath+initializationLockSuffix, os.O_CREATE|os.O_RDWR, 0o600)
200 if err != nil {
201 return nil, classify("open storage initialization lock", err)
202 }
203 for {
204 err = syscall.Flock(int(file.Fd()), syscall.LOCK_EX|syscall.LOCK_NB)
205 if err == nil {
206 return &initializationLock{file: file}, nil
207 }
208 if err != syscall.EWOULDBLOCK && err != syscall.EAGAIN && err != syscall.EINTR {
209 file.Close()
210 return nil, classify("acquire storage initialization lock", err)
211 }
212 select {
213 case <-ctx.Done():
214 file.Close()
215 return nil, fmt.Errorf("acquire storage initialization lock: %w", ctx.Err())
216 case <-time.After(10 * time.Millisecond):
217 }
218 }
219}
220
221func (lock *initializationLock) Close() error {
222 if lock == nil || lock.file == nil {
223 return nil
224 }
225 unlockErr := syscall.Flock(int(lock.file.Fd()), syscall.LOCK_UN)
226 closeErr := lock.file.Close()
227 if unlockErr != nil {
228 return classify("release storage initialization lock", unlockErr)
229 }
230 return classify("close storage initialization lock", closeErr)
231}
232
233func verifyPragmas(ctx context.Context, db *sql.DB) error {
234 var foreignKeys int
235 if err := db.QueryRowContext(ctx, `PRAGMA foreign_keys`).Scan(&foreignKeys); err != nil {
236 return classify("read foreign_keys pragma", err)
237 }
238 if foreignKeys != 1 {
239 return fmt.Errorf("foreign_keys pragma is %d, want 1: %w", foreignKeys, ErrSchemaInvalid)
240 }
241 var journalMode string
242 if err := db.QueryRowContext(ctx, `PRAGMA journal_mode`).Scan(&journalMode); err != nil {
243 return classify("read journal_mode pragma", err)
244 }
245 if !strings.EqualFold(journalMode, "wal") {
246 return fmt.Errorf("journal_mode pragma is %q, want wal: %w", journalMode, ErrSchemaInvalid)
247 }
248 var busyTimeout int
249 if err := db.QueryRowContext(ctx, `PRAGMA busy_timeout`).Scan(&busyTimeout); err != nil {
250 return classify("read busy_timeout pragma", err)
251 }
252 if busyTimeout != busyTimeoutMillis {
253 return fmt.Errorf("busy_timeout pragma is %d, want %d: %w", busyTimeout, busyTimeoutMillis, ErrSchemaInvalid)
254 }
255 return nil
256}
257
258func (s *Store) Close() error {
259 if s == nil {
260 return nil
261 }
262 s.closeOnce.Do(func() {
263 s.closed.Store(true)
264 s.attachMu.Lock()
265 attached := s.attached
266 s.attached = nil
267 s.attachMu.Unlock()
268 var errs []error
269 for i := len(attached) - 1; i >= 0; i-- {
270 errs = append(errs, attached[i].Close())
271 }
272 if s.sql != nil {
273 errs = append(errs, s.sql.Close())
274 }
275 s.closeErr = errors.Join(errs...)
276 })
277 return s.closeErr
278}
279
280func (s *Store) checkOpen(operation string) error {
281 if s == nil || s.closed.Load() {
282 return fmt.Errorf("%s: %w", operation, ErrClosed)
283 }
284 return nil
285}