Something went wrong. Try again.
An ATProto Powered Reputation Network
Something went wrong. Try again.
13 kB · 456 lines
Go
at master
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457package storage
import ( "context" "database/sql" _ "embed" "errors" "fmt" "log/slog" "os" "path/filepath" "time"
_ "github.com/mattn/go-sqlite3"
"provenance/internal/domain" "provenance/internal/storage/db" "provenance/internal/telemetry")
//go:embed schema.sqlvar schema string
//go:embed migration2.sqlvar migration2 string
//go:embed migration3.sqlvar migration3 string
//go:embed migration4.sqlvar migration4 string
//go:embed migration5.sqlvar migration5 string
//go:embed soft_schema.sqlvar softSchema string
//go:embed lists_schema.sqlvar listsSchema string
//go:embed migration7.sqlvar migration7 string
//go:embed legacy7.sqlvar legacy7 string
//go:embed migration8.sqlvar migration8 string
//go:embed migration8_copy.sqlvar migration8Copy string
//go:embed replay_schema.sqlvar replaySchema string
//go:embed migration10.sqlvar migration10 string
//go:embed migration11.sqlvar migration11 string
//go:embed soft_indexes.sqlvar softIndexes string
const SchemaVersion = 12
const sqliteDriver = "sqlite3"const sqliteCacheKiB = 16 * 1024 * 1024const sqliteBusyTimeoutMS = 5000const sqliteWALPages = 1000const sqliteReplayWALPages = 64 * 1024
type SQLite struct { path string conn *sql.DB queries *db.Queries softMode domain.SoftMode}
func Open(path string) (*SQLite, error) { return OpenWithSoft(path, domain.SoftDisabled)}
func OpenWithSoft(path string, mode domain.SoftMode) (*SQLite, error) { return openSQLite(path, mode, ReplayIndexesReady)}
// Full replay keeps an interrupted archive's indexes deferred until cutover.func OpenForReplay(path string, mode domain.SoftMode) (*SQLite, error) { return openSQLite(path, mode, ReplayIndexesDeferred)}
func openSQLite(path string, mode domain.SoftMode, indexes ReplayIndexMode) (store *SQLite, err error) { ctx, end := telemetry.Start(context.Background(), "storage.open", slog.LevelInfo) defer func() { end(err) }() if mode < domain.SoftDisabled || mode > domain.SoftAllAuthorLists { return nil, errors.New("invalid soft vouch mode") } if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil { return nil, err } file, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0600) if err != nil { return nil, err } if err := file.Chmod(0600); err != nil { file.Close() return nil, err } if err := file.Close(); err != nil { return nil, err } conn, err := sql.Open(sqliteDriver, path) if err != nil { return nil, err } conn.SetMaxOpenConns(1) // Cache database pages in memory to reduce repeated filesystem reads. for _, statement := range []string{ "PRAGMA journal_mode=WAL", fmt.Sprintf("PRAGMA busy_timeout=%d", sqliteBusyTimeoutMS), "PRAGMA synchronous=NORMAL", fmt.Sprintf("PRAGMA wal_autocheckpoint=%d", sqliteWALPages), fmt.Sprintf("PRAGMA cache_size=-%d", sqliteCacheKiB), "PRAGMA foreign_keys=ON", "PRAGMA temp_store=MEMORY", } { if _, err := conn.Exec(statement); err != nil { conn.Close() return nil, err } } if err := migrateContext(ctx, conn); err != nil { conn.Close() return nil, err } // Reuse SQL statements so that replay does not parse each query for every record. queries, err := db.Prepare(context.Background(), conn) if err != nil { conn.Close() return nil, err } s := &SQLite{conn: conn, queries: queries, softMode: mode, path: path} if indexes == ReplayIndexesReady { if err := s.FinishReplay(ctx); err != nil { s.Close() return nil, err } } return s, nil}
// Apply schema upgrades atomically and reject databases from a newer release.func migrate(conn *sql.DB) error { return migrateContext(context.Background(), conn)}
func migrateContext(ctx context.Context, conn *sql.DB) (err error) { _, end := telemetry.Start(ctx, "storage.migrate", slog.LevelInfo) defer func() { end(err) }() tx, err := conn.Begin() if err != nil { return err } defer tx.Rollback() var version int if err := tx.QueryRow("PRAGMA user_version").Scan(&version); err != nil { return err } if version > SchemaVersion { return errors.New("database schema is newer than this service") } if version == SchemaVersion { return tx.Commit() } if version >= 8 { for _, statement := range []string{migration10, schema, softSchema, listsSchema, replaySchema} { if _, err := tx.Exec(statement); err != nil { return err } } if err := migrateTrustCache(tx); err != nil { return err } if err := migrateReplayIndexes(tx); err != nil { return err } if _, err := tx.Exec(fmt.Sprintf("PRAGMA user_version=%d", SchemaVersion)); err != nil { return err } return tx.Commit() } if version == 1 { if _, err := tx.Exec(migration2); err != nil { return err } q := db.New(tx) rows, err := q.LegacyKeys(context.Background()) if err != nil { return err } for _, row := range rows { if err := q.SetLegacyKey(context.Background(), db.SetLegacyKeyParams{Author: row.Author, Subject: row.Subject, Tag: row.Tag, RecordKey: domain.RecordKey(row.Subject, row.Tag)}); err != nil { return err } } version = 2 } if version == 2 { if _, err := tx.Exec(migration3); err != nil { return err } } // Older schemas first acquire cache tables, then upgrade their dependency identities. if version > 0 && version < 4 { if _, err := tx.Exec(migration4); err != nil { return err } } if version > 0 && version < 5 { if _, err := tx.Exec(migration5); err != nil { return err } } if version > 0 { if _, err := tx.Exec(legacy7); err != nil { return err } if _, err := tx.Exec(migration7); err != nil { return err } if _, err := tx.Exec(migration8); err != nil { return err } if _, err := tx.Exec(migration10); err != nil { return err } } if _, err := tx.Exec(schema); err != nil { return err } if _, err := tx.Exec(softSchema); err != nil { return err } if _, err := tx.Exec(listsSchema); err != nil { return err } if _, err := tx.Exec(replaySchema); err != nil { return err } if version > 0 { if _, err := tx.Exec(migration8Copy); err != nil { return err } if err := migrateTrustCache(tx); err != nil { return err } } if err := migrateReplayIndexes(tx); err != nil { return err } if _, err := tx.Exec(fmt.Sprintf("PRAGMA user_version=%d", SchemaVersion)); err != nil { return err } return tx.Commit()}
func migrateReplayIndexes(tx *sql.Tx) error { var deferred int if err := tx.QueryRow("SELECT deferred FROM replay_indexes WHERE singleton = 1").Scan(&deferred); err != nil { return err } if deferred != 0 { return nil } _, err := tx.Exec(softIndexes) return err}
func migrateTrustCache(tx *sql.Tx) error { var legacy int if err := tx.QueryRow("SELECT count(*) FROM pragma_table_info('trust_cache') WHERE name = 'result'").Scan(&legacy); err != nil { return err } if legacy == 0 { return nil } _, err := tx.Exec(migration11) return err}
func (s *SQLite) Close() error { return errors.Join(s.queries.Close(), s.conn.Close()) }
func (s *SQLite) Join(ctx context.Context, did string) error { return s.queries.Join(ctx, did) }
func (s *SQLite) Participant(ctx context.Context, did string) (Participant, error) { p, err := s.queries.Participant(ctx, did) return participant(p), err}
func (s *SQLite) Participants(ctx context.Context, limit int) ([]Participant, error) { rows, err := s.queries.Participants(ctx, int64(limit)) if err != nil { return nil, err } result := make([]Participant, 0, len(rows)) for _, p := range rows { result = append(result, participant(p)) } return result, nil}
func participant(p db.Participant) Participant { return Participant{DID: p.Did, Threshold: p.Threshold, SyncedAt: p.SyncedAt, BootstrapUS: p.BootstrapUs, Active: p.Active != 0}}
func (s *SQLite) SetThreshold(ctx context.Context, did string, value float64) error { if err := domain.ValidateThreshold(value); err != nil { return err } count, err := s.queries.SetThreshold(ctx, db.SetThresholdParams{Threshold: value, Did: did}) if err != nil { return err } if count == 0 { return sql.ErrNoRows } return nil}
func (s *SQLite) Replace(ctx context.Context, did string, vouches []domain.Vouch, at time.Time) error { return s.replace(ctx, did, vouches, nil, at)}
func (s *SQLite) ReplaceWithLists(ctx context.Context, did string, vouches []domain.Vouch, lists domain.ListSnapshot, at time.Time) error { if !domain.SoftLists(s.softMode) { return errors.New("soft lists are disabled") } return s.replace(ctx, did, vouches, &lists, at)}
func (s *SQLite) replace(ctx context.Context, did string, vouches []domain.Vouch, lists *domain.ListSnapshot, at time.Time) error { tx, err := s.conn.BeginTx(ctx, nil) if err != nil { return err } defer tx.Rollback() q := s.queries.WithTx(tx) if lists != nil { if err := replaceLists(ctx, q, did, *lists, at); err != nil { return err } } previous, err := q.AuthorRecords(ctx, did) if err != nil { return err } keys := make(map[string]struct{}, len(vouches)) if err := q.ClearVersions(ctx, did); err != nil { return err } for _, v := range vouches { if domain.SoftSource(v.Source) { if s.softMode == domain.SoftDisabled { return errors.New("soft snapshots are disabled") } continue } if v.Source != "" && v.Source != domain.Collection { return errors.New("invalid explicit vouch source") } if v.Author != did { return errors.New("vouch author does not match repository") } if err := domain.ValidateScore(v.Score); err != nil { return err } if err := domain.ValidateTag(v.Tag); err != nil { return err } if err := domain.ValidateComment(v.Comment); err != nil { return err } key := domain.RecordKey(v.Subject, v.Tag) if _, exists := keys[key]; exists { return errors.New("snapshot contains duplicate records") } keys[key] = struct{}{} if err := q.ReplaceVouch(ctx, db.ReplaceVouchParams{Author: did, Subject: v.Subject, Tag: v.Tag, Score: int64(v.Score), CreatedAt: v.CreatedAt, RecordKey: key, Comment: v.Comment}); err != nil { return err } } for _, row := range previous { if _, exists := keys[row.RecordKey]; exists { continue } if err := q.DeleteRecord(ctx, db.DeleteRecordParams{Author: did, RecordKey: row.RecordKey}); err != nil { return err } } if s.softMode != domain.SoftDisabled { if err := replaceSoft(ctx, q, did, vouches, at); err != nil { return err } } if err := q.SetBootstrap(ctx, db.SetBootstrapParams{Did: did, SyncedAt: at.UTC().Format(time.RFC3339Nano), BootstrapUs: at.UnixMicro(), AccountUs: at.UnixMicro()}); err != nil { return err } return tx.Commit()}
func (s *SQLite) Graph(ctx context.Context, tag string, mode domain.Mode, limit int) ([]domain.Vouch, error) { return s.graphRows(ctx, s.queries, tag, mode, limit)}
func (s *SQLite) OwnVouches(ctx context.Context, did string, limit, offset int) ([]domain.Vouch, error) { rows, err := s.queries.OwnVouches(ctx, db.OwnVouchesParams{Author: did, Limit: int64(limit), Offset: int64(offset)}) return vouchRows(rows), err}
func (s *SQLite) Tags(ctx context.Context, prefix string, limit int) ([]string, error) { return s.queries.Tags(ctx, db.TagsParams{Prefix: prefix, MaxRows: int64(limit)})}
func vouchRows(rows []db.Vouch) []domain.Vouch { result := make([]domain.Vouch, 0, len(rows)) for _, v := range rows { result = append(result, domain.Vouch{Author: v.Author, Subject: v.Subject, Tag: v.Tag, Score: float64(v.Score), CreatedAt: v.CreatedAt, Comment: v.Comment, Source: v.Source, RecordKey: v.RecordKey}) } return result}
func (s *SQLite) Put(ctx context.Context, kind, id string, value []byte, expires time.Time) error { return s.queries.PutSecret(ctx, db.PutSecretParams{Kind: kind, ID: id, Value: value, Expires: expires.Unix()})}
func (s *SQLite) Add(ctx context.Context, kind, id string, value []byte, expires time.Time) error { return s.queries.AddSecret(ctx, db.AddSecretParams{Kind: kind, ID: id, Value: value, Expires: expires.Unix()})}
func (s *SQLite) Get(ctx context.Context, kind, id string) ([]byte, error) { return s.queries.GetSecret(ctx, db.GetSecretParams{Kind: kind, ID: id, Expires: time.Now().Unix()})}
func (s *SQLite) Consume(ctx context.Context, kind, id string) ([]byte, error) { return s.queries.ConsumeSecret(ctx, db.ConsumeSecretParams{Kind: kind, ID: id, Expires: time.Now().Unix()})}
func (s *SQLite) Remove(ctx context.Context, kind, id string) error { return s.queries.RemoveSecret(ctx, db.RemoveSecretParams{Kind: kind, ID: id})}
func (s *SQLite) Prune(ctx context.Context) error { return s.queries.PruneSecrets(ctx, time.Now().Unix())}