basedb.go 6.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249
  1. // Copyright (C) 2025 The Syncthing Authors.
  2. //
  3. // This Source Code Form is subject to the terms of the Mozilla Public
  4. // License, v. 2.0. If a copy of the MPL was not distributed with this file,
  5. // You can obtain one at https://mozilla.org/MPL/2.0/.
  6. package sqlite
  7. import (
  8. "database/sql"
  9. "embed"
  10. "io/fs"
  11. "path/filepath"
  12. "strconv"
  13. "strings"
  14. "sync"
  15. "text/template"
  16. "time"
  17. "github.com/jmoiron/sqlx"
  18. "github.com/syncthing/syncthing/lib/build"
  19. "github.com/syncthing/syncthing/lib/protocol"
  20. )
  21. const currentSchemaVersion = 1
  22. //go:embed sql/**
  23. var embedded embed.FS
  24. type baseDB struct {
  25. path string
  26. baseName string
  27. sql *sqlx.DB
  28. updateLock sync.Mutex
  29. updatePoints int
  30. checkpointsCount int
  31. statementsMut sync.RWMutex
  32. statements map[string]*sqlx.Stmt
  33. tplInput map[string]any
  34. }
  35. func openBase(path string, maxConns int, pragmas, schemaScripts, migrationScripts []string) (*baseDB, error) {
  36. // Open the database with options to enable foreign keys and recursive
  37. // triggers (needed for the delete+insert triggers on row replace).
  38. sqlDB, err := sqlx.Open(dbDriver, "file:"+path+"?"+commonOptions)
  39. if err != nil {
  40. return nil, wrap(err)
  41. }
  42. sqlDB.SetMaxOpenConns(maxConns)
  43. for _, pragma := range pragmas {
  44. if _, err := sqlDB.Exec("PRAGMA " + pragma); err != nil {
  45. return nil, wrap(err, "PRAGMA "+pragma)
  46. }
  47. }
  48. db := &baseDB{
  49. path: path,
  50. baseName: filepath.Base(path),
  51. sql: sqlDB,
  52. statements: make(map[string]*sqlx.Stmt),
  53. }
  54. for _, script := range schemaScripts {
  55. if err := db.runScripts(script); err != nil {
  56. return nil, wrap(err)
  57. }
  58. }
  59. ver, _ := db.getAppliedSchemaVersion()
  60. if ver.SchemaVersion > 0 {
  61. filter := func(scr string) bool {
  62. scr = filepath.Base(scr)
  63. nstr, _, ok := strings.Cut(scr, "-")
  64. if !ok {
  65. return false
  66. }
  67. n, err := strconv.ParseInt(nstr, 10, 32)
  68. if err != nil {
  69. return false
  70. }
  71. return int(n) > ver.SchemaVersion
  72. }
  73. for _, script := range migrationScripts {
  74. if err := db.runScripts(script, filter); err != nil {
  75. return nil, wrap(err)
  76. }
  77. }
  78. }
  79. // Set the current schema version, if not already set
  80. if err := db.setAppliedSchemaVersion(currentSchemaVersion); err != nil {
  81. return nil, wrap(err)
  82. }
  83. db.tplInput = map[string]any{
  84. "FlagLocalUnsupported": protocol.FlagLocalUnsupported,
  85. "FlagLocalIgnored": protocol.FlagLocalIgnored,
  86. "FlagLocalMustRescan": protocol.FlagLocalMustRescan,
  87. "FlagLocalReceiveOnly": protocol.FlagLocalReceiveOnly,
  88. "FlagLocalGlobal": protocol.FlagLocalGlobal,
  89. "FlagLocalNeeded": protocol.FlagLocalNeeded,
  90. "SyncthingVersion": build.LongVersion,
  91. }
  92. return db, nil
  93. }
  94. func (s *baseDB) Close() error {
  95. s.updateLock.Lock()
  96. s.statementsMut.Lock()
  97. defer s.updateLock.Unlock()
  98. defer s.statementsMut.Unlock()
  99. for _, stmt := range s.statements {
  100. stmt.Close()
  101. }
  102. return wrap(s.sql.Close())
  103. }
  104. var tplFuncs = template.FuncMap{
  105. "or": func(vs ...int) int {
  106. v := vs[0]
  107. for _, ov := range vs[1:] {
  108. v |= ov
  109. }
  110. return v
  111. },
  112. }
  113. // stmt returns a prepared statement for the given SQL string, after
  114. // applying local template expansions. The statement is cached.
  115. func (s *baseDB) stmt(tpl string) stmt {
  116. tpl = strings.TrimSpace(tpl)
  117. // Fast concurrent lookup of cached statement
  118. s.statementsMut.RLock()
  119. stmt, ok := s.statements[tpl]
  120. s.statementsMut.RUnlock()
  121. if ok {
  122. return stmt
  123. }
  124. // On miss, take the full lock, check again
  125. s.statementsMut.Lock()
  126. defer s.statementsMut.Unlock()
  127. stmt, ok = s.statements[tpl]
  128. if ok {
  129. return stmt
  130. }
  131. // Apply template expansions
  132. var sb strings.Builder
  133. compTpl := template.Must(template.New("tpl").Funcs(tplFuncs).Parse(tpl))
  134. if err := compTpl.Execute(&sb, s.tplInput); err != nil {
  135. panic("bug: bad template: " + err.Error())
  136. }
  137. // Prepare and cache
  138. stmt, err := s.sql.Preparex(sb.String())
  139. if err != nil {
  140. return failedStmt{err}
  141. }
  142. s.statements[tpl] = stmt
  143. return stmt
  144. }
  145. type stmt interface {
  146. Exec(args ...any) (sql.Result, error)
  147. Get(dest any, args ...any) error
  148. Queryx(args ...any) (*sqlx.Rows, error)
  149. Select(dest any, args ...any) error
  150. }
  151. type failedStmt struct {
  152. err error
  153. }
  154. func (f failedStmt) Exec(_ ...any) (sql.Result, error) { return nil, f.err }
  155. func (f failedStmt) Get(_ any, _ ...any) error { return f.err }
  156. func (f failedStmt) Queryx(_ ...any) (*sqlx.Rows, error) { return nil, f.err }
  157. func (f failedStmt) Select(_ any, _ ...any) error { return f.err }
  158. func (s *baseDB) runScripts(glob string, filter ...func(s string) bool) error {
  159. scripts, err := fs.Glob(embedded, glob)
  160. if err != nil {
  161. return wrap(err)
  162. }
  163. tx, err := s.sql.Begin()
  164. if err != nil {
  165. return wrap(err)
  166. }
  167. defer tx.Rollback() //nolint:errcheck
  168. nextScript:
  169. for _, scr := range scripts {
  170. for _, fn := range filter {
  171. if !fn(scr) {
  172. continue nextScript
  173. }
  174. }
  175. bs, err := fs.ReadFile(embedded, scr)
  176. if err != nil {
  177. return wrap(err, scr)
  178. }
  179. // SQLite requires one statement per exec, so we split the init
  180. // files on lines containing only a semicolon and execute them
  181. // separately. We require it on a separate line because there are
  182. // also statement-internal semicolons in the triggers.
  183. for _, stmt := range strings.Split(string(bs), "\n;") {
  184. if _, err := tx.Exec(stmt); err != nil {
  185. return wrap(err, stmt)
  186. }
  187. }
  188. }
  189. return wrap(tx.Commit())
  190. }
  191. type schemaVersion struct {
  192. SchemaVersion int
  193. AppliedAt int64
  194. SyncthingVersion string
  195. }
  196. func (s *schemaVersion) AppliedTime() time.Time {
  197. return time.Unix(0, s.AppliedAt)
  198. }
  199. func (s *baseDB) setAppliedSchemaVersion(ver int) error {
  200. _, err := s.stmt(`
  201. INSERT OR IGNORE INTO schemamigrations (schema_version, applied_at, syncthing_version)
  202. VALUES (?, ?, ?)
  203. `).Exec(ver, time.Now().UnixNano(), build.LongVersion)
  204. return wrap(err)
  205. }
  206. func (s *baseDB) getAppliedSchemaVersion() (schemaVersion, error) {
  207. var v schemaVersion
  208. err := s.stmt(`
  209. SELECT schema_version as schemaversion, applied_at as appliedat, syncthing_version as syncthingversion FROM schemamigrations
  210. ORDER BY schema_version DESC
  211. LIMIT 1
  212. `).Get(&v)
  213. return v, wrap(err)
  214. }