Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
5.2 kB · 217 lines
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218package db
import ( "context" "database/sql" "errors" "fmt"
"github.com/bluesky-social/indigo/atproto/syntax" "tangled.org/core/knotmirror/models")
func BufferEvent(ctx context.Context, e *sql.DB, did syntax.DID, rkey syntax.RecordKey, eventRkey string) error { tx, err := e.BeginTx(ctx, nil) if err != nil { return fmt.Errorf("begin tx: %w", err) } defer tx.Rollback()
var state models.RepoState if err := tx.QueryRowContext(ctx, `select state from repos where did = $1 and rkey = $2 for update`, did, rkey, ).Scan(&state); err != nil { if errors.Is(err, sql.ErrNoRows) { return nil } return fmt.Errorf("locking repo: %w", err) } if state != models.RepoStateResyncing { return nil }
if _, err := tx.ExecContext(ctx, `insert into resync_buffer (did, rkey, event_rkey) values ($1, $2, $3) on conflict (did, rkey, event_rkey) do nothing`, did, rkey, eventRkey, ); err != nil { return fmt.Errorf("buffering event: %w", err) }
if err := tx.Commit(); err != nil { return fmt.Errorf("commit: %w", err) } return nil}
func MarkDesync(ctx context.Context, e *sql.DB, did syntax.DID, rkey syntax.RecordKey) error { tx, err := e.BeginTx(ctx, nil) if err != nil { return fmt.Errorf("begin tx: %w", err) } defer tx.Rollback()
var state models.RepoState if err := tx.QueryRowContext(ctx, `select state from repos where did = $1 and rkey = $2 for update`, did, rkey, ).Scan(&state); err != nil { if errors.Is(err, sql.ErrNoRows) { return nil } return fmt.Errorf("locking repo: %w", err) } switch state { case models.RepoStateActive, models.RepoStateDesynchronized, models.RepoStateError: default: return nil }
if _, err := tx.ExecContext(ctx, `update repos set state = $1 where did = $2 and rkey = $3`, models.RepoStateDesynchronized, did, rkey, ); err != nil { return fmt.Errorf("marking desync: %w", err) }
if err := tx.Commit(); err != nil { return fmt.Errorf("commit: %w", err) } return nil}
func CompleteResync(ctx context.Context, e *sql.DB, did syntax.DID, rkey syntax.RecordKey) error { tx, err := e.BeginTx(ctx, nil) if err != nil { return fmt.Errorf("begin tx: %w", err) } defer tx.Rollback()
var state models.RepoState if err := tx.QueryRowContext(ctx, `select state from repos where did = $1 and rkey = $2 for update`, did, rkey, ).Scan(&state); err != nil { if errors.Is(err, sql.ErrNoRows) { return nil } return fmt.Errorf("locking repo: %w", err) } if state != models.RepoStateResyncing { return nil }
res, err := tx.ExecContext(ctx, `delete from resync_buffer where did = $1 and rkey = $2`, did, rkey, ) if err != nil { return fmt.Errorf("draining buffer: %w", err) } deleted, _ := res.RowsAffected()
finalState := models.RepoStateActive if deleted > 0 { finalState = models.RepoStateDesynchronized }
if _, err := tx.ExecContext(ctx, `update repos set error_msg = '', retry_count = 0, retry_after = 0, state = $1 where did = $2 and rkey = $3`, finalState, did, rkey, ); err != nil { return fmt.Errorf("finalising repo: %w", err) }
return tx.Commit()}
func CancelResync(ctx context.Context, e *sql.DB, aturi syntax.ATURI) error { tx, err := e.BeginTx(ctx, nil) if err != nil { return fmt.Errorf("begin tx: %w", err) } defer tx.Rollback()
var did syntax.DID var rkey syntax.RecordKey var state models.RepoState if err := tx.QueryRowContext(ctx, `select did, rkey, state from repos where at_uri = $1 for update`, aturi, ).Scan(&did, &rkey, &state); err != nil { if errors.Is(err, sql.ErrNoRows) { return nil } return fmt.Errorf("locking repo: %w", err) } if state != models.RepoStateResyncing { return nil }
if _, err := tx.ExecContext(ctx, `delete from resync_buffer where did = $1 and rkey = $2`, did, rkey, ); err != nil { return fmt.Errorf("draining buffer: %w", err) }
if _, err := tx.ExecContext(ctx, `update repos set state = $1 where did = $2 and rkey = $3`, models.RepoStateSuspended, did, rkey, ); err != nil { return fmt.Errorf("suspending repo: %w", err) }
return tx.Commit()}
func FailResync(ctx context.Context, e *sql.DB, did syntax.DID, rkey syntax.RecordKey, state models.RepoState, errMsg string, retryCount int, retryAfter int64) error { tx, err := e.BeginTx(ctx, nil) if err != nil { return fmt.Errorf("begin tx: %w", err) } defer tx.Rollback()
var current models.RepoState if err := tx.QueryRowContext(ctx, `select state from repos where did = $1 and rkey = $2 for update`, did, rkey, ).Scan(¤t); err != nil { if errors.Is(err, sql.ErrNoRows) { return nil } return fmt.Errorf("locking repo: %w", err) } if current != models.RepoStateResyncing { return nil }
if _, err := tx.ExecContext(ctx, `delete from resync_buffer where did = $1 and rkey = $2`, did, rkey, ); err != nil { return fmt.Errorf("draining buffer: %w", err) }
if _, err := tx.ExecContext(ctx, `update repos set error_msg = $1, retry_count = $2, retry_after = $3, state = $4 where did = $5 and rkey = $6`, errMsg, retryCount, retryAfter, state, did, rkey, ); err != nil { return fmt.Errorf("finalising repo: %w", err) }
return tx.Commit()}