Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125package db
import ( "database/sql" "time"
"github.com/bluesky-social/indigo/atproto/syntax")
type ParkedRecord struct { URI string Rev string Event []byte Attempts int RetryAt time.Time Err string}
func (d *DB) ParkRecord(p ParkedRecord) error { _, err := d.Exec( `insert into backfill_parked (uri, rev, event, attempts, retry_at, error) values (?, ?, ?, 0, ?, ?) on conflict(uri) do update set rev = excluded.rev, event = excluded.event, attempts = 0, retry_at = excluded.retry_at, error = excluded.error where excluded.rev >= backfill_parked.rev`, p.URI, p.Rev, p.Event, p.RetryAt.Unix(), p.Err, ) return err}
func (d *DB) DueParkedRecords(before time.Time) ([]ParkedRecord, error) { rows, err := d.Query( `select uri, rev, event, attempts, retry_at, error from backfill_parked where retry_at <= ? order by retry_at`, before.Unix(), ) if err != nil { return nil, err } defer rows.Close() var out []ParkedRecord for rows.Next() { var p ParkedRecord var retryAt int64 if err := rows.Scan(&p.URI, &p.Rev, &p.Event, &p.Attempts, &retryAt, &p.Err); err != nil { return nil, err } p.RetryAt = time.Unix(retryAt, 0) out = append(out, p) } return out, rows.Err()}
func (d *DB) NextParkedRetry() (time.Time, error) { var at sql.NullInt64 if err := d.QueryRow(`select min(retry_at) from backfill_parked`).Scan(&at); err != nil { return time.Time{}, err } if !at.Valid { return time.Time{}, nil } return time.Unix(at.Int64, 0), nil}
func (d *DB) RescheduleParked(uri, rev string, attempts int, retryAt time.Time, cause string) error { _, err := d.Exec( `update backfill_parked set attempts = ?, retry_at = ?, error = ? where uri = ? and rev = ?`, attempts, retryAt.Unix(), cause, uri, rev, ) return err}
func (d *DB) UnparkRecord(uri, rev string) error { _, err := d.Exec(`delete from backfill_parked where uri = ? and rev = ?`, uri, rev) return err}
func (d *DB) CountParked() (int, error) { var n int err := d.QueryRow(`select count(*) from backfill_parked`).Scan(&n) return n, err}
// AddPendingCatchUp keeps the first since, so a did that never backfills still ages outfunc (d *DB) AddPendingCatchUp(dids []syntax.DID, since time.Time) error { tx, err := d.Begin() if err != nil { return err } defer tx.Rollback() for _, did := range dids { if _, err := tx.Exec(`insert or ignore into backfill_pending (did, since) values (?, ?)`, did, since.Unix()); err != nil { return err } } return tx.Commit()}
func (d *DB) PendingCatchUp() ([]syntax.DID, error) { return queryDIDs(d, `select did from backfill_pending order by did`)}
func (d *DB) DeletePendingCatchUp(dids ...syntax.DID) error { tx, err := d.Begin() if err != nil { return err } defer tx.Rollback() for _, did := range dids { if _, err := tx.Exec(`delete from backfill_pending where did = ?`, did); err != nil { return err } } return tx.Commit()}
func (d *DB) DropStalePendingCatchUp(olderThan time.Time) ([]syntax.DID, error) { return queryDIDs(d, `delete from backfill_pending where since < ? returning did`, olderThan.Unix())}