Something went wrong. Try again.
An ATProto Powered Reputation Network
Something went wrong. Try again.
5.7 kB · 176 lines
Go
at master
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177package storage
import ( "context" "database/sql" "errors" "math" "time"
"provenance/internal/domain" "provenance/internal/storage/db" "provenance/internal/trust")
// Read the graph and its revision in one snapshot for safe cache publication.func (s *SQLite) TrustGraph(ctx context.Context, tag string, mode domain.Mode, limit int) ([]domain.Vouch, int64, error) { tx, err := s.conn.BeginTx(ctx, nil) if err != nil { return nil, 0, err } defer tx.Rollback() q := s.queries.WithTx(tx) rows, err := s.graphRows(ctx, q, tag, mode, limit) if err != nil { return nil, 0, err } revision, err := q.GraphRevision(ctx) if err != nil { return nil, 0, err } if err := tx.Commit(); err != nil { return nil, 0, err } return rows, revision, nil}
func (s *SQLite) Incoming(ctx context.Context, target, tag string, mode domain.Mode, limit int) ([]domain.Vouch, error) { fallback := int64(0) if mode == domain.Fallback { fallback = 1 } rows, err := s.queries.Incoming(ctx, db.IncomingParams{Subject: target, Tag: tag, Column3: fallback, Limit: int64(limit)}) if err != nil { return nil, err } result := vouchRows(rows) if s.softMode == domain.SoftDisabled || mode == domain.Tagged { return result, nil } all := int64(0) if domain.SoftAll(s.softMode) { all = 1 } soft, err := s.queries.SoftIncoming(ctx, db.SoftIncomingParams{Subject: target, Tag: tag, Fallback: fallback, AllAuthors: all, MaxRows: int64(limit)}) if err != nil { return nil, err } result = appendSoftRows(result, soft) if !domain.SoftLists(s.softMode) { return result, nil } lists, err := s.queries.ListIncoming(ctx, db.ListIncomingParams{Subject: target, Tag: tag, Fallback: fallback, AllAuthors: all, MaxRows: int64(limit)}) return appendListRows(result, lists), err}
func (s *SQLite) CachedTrust(ctx context.Context, key string, now time.Time) (trust.Result, error) { // Read the values and ordered children in one snapshot so invalidation cannot split the result. tx, err := s.conn.BeginTx(ctx, nil) if err != nil { return trust.Result{}, err } defer tx.Rollback() q := s.queries.WithTx(tx) row, err := q.CachedTrust(ctx, db.CachedTrustParams{CacheKey: key, ExpiresUs: now.UnixMicro()}) if err != nil { return trust.Result{}, err } result := trust.Result{Status: trust.Status(row.Status), Score: scorePointer(row.Score)} path, err := q.TrustCachePath(ctx, key) if err != nil { return trust.Result{}, err } if len(path) > 0 { result.Path = path } scores, err := q.TrustCacheScores(ctx, key) if err != nil { return trust.Result{}, err } for _, score := range scores { result.PathScores = append(result.PathScores, trust.PathScore{DID: score.Did, Assertion: scorePointer(score.Assertion), Source: score.Source, ListURI: score.ListUri, Score: score.Score, Comment: score.Comment}) } if err := tx.Commit(); err != nil { return trust.Result{}, err } return result, nil}
// The revision check and dependency writes share a transaction with cache publication.func (s *SQLite) SaveTrust(ctx context.Context, key string, result trust.Result, dependencies []trust.Dependency, revision int64, expires time.Time) (bool, error) { if err := validateCacheScores(result); err != nil { return false, err } tx, err := s.conn.BeginTx(ctx, nil) if err != nil { return false, err } defer tx.Rollback() q := s.queries.WithTx(tx) current, err := q.GraphRevision(ctx) if err != nil { return false, err } if current != revision { return false, nil } if err := q.DeleteTrust(ctx, key); err != nil { return false, err } if err := q.SaveTrust(ctx, db.SaveTrustParams{CacheKey: key, Status: string(result.Status), Score: nullableScore(result.Score), ExpiresUs: expires.UnixMicro()}); err != nil { return false, err } for position, did := range result.Path { if err := q.SaveTrustPath(ctx, db.SaveTrustPathParams{CacheKey: key, Position: int64(position), Did: did}); err != nil { return false, err } } for position, score := range result.PathScores { if err := q.SaveTrustScore(ctx, db.SaveTrustScoreParams{CacheKey: key, Position: int64(position), Did: score.DID, Assertion: nullableScore(score.Assertion), Source: score.Source, ListUri: score.ListURI, Score: score.Score, Comment: score.Comment}); err != nil { return false, err } } for _, dependency := range dependencies { if err := q.PutIdentity(ctx, dependency.Author); err != nil { return false, err } source := dependency.Source if source == "" { source = domain.Collection } if err := q.SaveDependency(ctx, db.SaveDependencyParams{CacheKey: key, Author: dependency.Author, Source: source, RecordKey: dependency.RecordKey}); err != nil { return false, err } } return true, tx.Commit()}
func nullableScore(value *float64) sql.NullFloat64 { if value == nil { return sql.NullFloat64{} } return sql.NullFloat64{Float64: *value, Valid: true}}
func scorePointer(value sql.NullFloat64) *float64 { if !value.Valid { return nil } return &value.Float64}
// Reject nonfinite values rather than let SQLite convert NaN to an unknown score.func validateCacheScores(result trust.Result) error { if result.Score != nil && (math.IsNaN(*result.Score) || math.IsInf(*result.Score, 0)) { return errors.New("trust cache score must be finite") } for _, score := range result.PathScores { if math.IsNaN(score.Score) || math.IsInf(score.Score, 0) { return errors.New("trust cache path score must be finite") } if score.Assertion != nil && (math.IsNaN(*score.Assertion) || math.IsInf(*score.Assertion, 0)) { return errors.New("trust cache assertion must be finite") } } return nil}