package 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 out
func (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())
}