282 lines
9.1 KiB
Go
282 lines
9.1 KiB
Go
package outbox
|
|
|
|
import (
|
|
"context"
|
|
"crypto/rand"
|
|
"database/sql"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"os"
|
|
"path/filepath"
|
|
"time"
|
|
|
|
_ "modernc.org/sqlite"
|
|
)
|
|
|
|
var ErrConflict = errors.New("idempotency key already used for different content")
|
|
var ErrNotRetryable = errors.New("only dead deliveries can be retried")
|
|
|
|
type Store struct{ db *sql.DB }
|
|
type Job struct {
|
|
ID string `json:"id"`
|
|
ReceiptID string `json:"receipt_id"`
|
|
MappingID string `json:"mapping_id"`
|
|
Target string `json:"target"`
|
|
State string `json:"state"`
|
|
Attempts int `json:"attempts"`
|
|
Created int64 `json:"created_at"`
|
|
Next int64 `json:"next_attempt_at"`
|
|
Status int `json:"status_code"`
|
|
Error string `json:"error,omitempty"`
|
|
Data json.RawMessage `json:"-"`
|
|
Lease string `json:"-"`
|
|
}
|
|
type Receipt struct {
|
|
ID string `json:"id"`
|
|
Duplicate bool `json:"duplicate"`
|
|
}
|
|
type Attempt struct {
|
|
At int64 `json:"at"`
|
|
Status int `json:"status_code"`
|
|
Error string `json:"error,omitempty"`
|
|
State string `json:"state"`
|
|
}
|
|
|
|
func ID() string {
|
|
b := make([]byte, 16)
|
|
if _, err := rand.Read(b); err != nil {
|
|
panic(err)
|
|
}
|
|
return hex.EncodeToString(b)
|
|
}
|
|
|
|
func Open(path string) (*Store, error) {
|
|
if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil {
|
|
return nil, err
|
|
}
|
|
// Create with restrictive permissions before SQLite opens the file.
|
|
f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0600)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
f.Close()
|
|
if err := os.Chmod(path, 0600); err != nil {
|
|
return nil, err
|
|
}
|
|
db, err := sql.Open("sqlite", path)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
db.SetMaxOpenConns(1)
|
|
s := &Store{db: db}
|
|
fail := func(err error) (*Store, error) { db.Close(); return nil, err }
|
|
for _, q := range []string{"PRAGMA busy_timeout=2000", "PRAGMA journal_mode=WAL", "PRAGMA synchronous=FULL", "PRAGMA foreign_keys=ON"} {
|
|
if _, err := db.Exec(q); err != nil {
|
|
return fail(err)
|
|
}
|
|
}
|
|
var version int
|
|
if err := db.QueryRow("PRAGMA user_version").Scan(&version); err != nil {
|
|
return fail(err)
|
|
}
|
|
if version > 1 {
|
|
return fail(fmt.Errorf("unsupported outbox schema %d", version))
|
|
}
|
|
if version == 0 {
|
|
tx, err := db.Begin()
|
|
if err != nil {
|
|
return fail(err)
|
|
}
|
|
_, err = tx.Exec(`CREATE TABLE receipts(id TEXT PRIMARY KEY,scope TEXT NOT NULL,key TEXT NOT NULL,digest TEXT NOT NULL,created INTEGER NOT NULL,UNIQUE(scope,key));
|
|
CREATE TABLE deliveries(id TEXT PRIMARY KEY,receipt_id TEXT NOT NULL REFERENCES receipts(id),mapping_id TEXT NOT NULL,target TEXT NOT NULL,data BLOB NOT NULL,state TEXT NOT NULL,attempts INTEGER NOT NULL DEFAULT 0,created INTEGER NOT NULL,next INTEGER NOT NULL,status INTEGER NOT NULL DEFAULT 0,error TEXT NOT NULL DEFAULT '',lease TEXT NOT NULL DEFAULT '');
|
|
CREATE INDEX due_deliveries ON deliveries(state,next);
|
|
CREATE TABLE attempts(id INTEGER PRIMARY KEY,delivery_id TEXT NOT NULL REFERENCES deliveries(id),at INTEGER NOT NULL,status INTEGER NOT NULL,error TEXT NOT NULL,state TEXT NOT NULL);
|
|
CREATE TABLE checkpoints(id TEXT PRIMARY KEY,value TEXT NOT NULL);
|
|
PRAGMA user_version=1;`)
|
|
if err != nil {
|
|
tx.Rollback()
|
|
return fail(err)
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return fail(err)
|
|
}
|
|
}
|
|
return s, nil
|
|
}
|
|
func (s *Store) Close() error { return s.db.Close() }
|
|
func (s *Store) Ping(ctx context.Context) error { return s.db.PingContext(ctx) }
|
|
|
|
func (s *Store) Lookup(ctx context.Context, scope, key, digest string) (Receipt, error) {
|
|
var r Receipt
|
|
var old string
|
|
err := s.db.QueryRowContext(ctx, "SELECT id,digest FROM receipts WHERE scope=? AND key=?", scope, key).Scan(&r.ID, &old)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return Receipt{}, nil
|
|
}
|
|
if err != nil {
|
|
return r, err
|
|
}
|
|
if old != digest {
|
|
return Receipt{}, ErrConflict
|
|
}
|
|
r.Duplicate = true
|
|
return r, nil
|
|
}
|
|
|
|
// Enqueue commits the receipt and every destination together, before acceptance.
|
|
func (s *Store) Enqueue(ctx context.Context, scope, key, digest string, jobs []Job) (Receipt, error) {
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return Receipt{}, err
|
|
}
|
|
defer tx.Rollback()
|
|
r := Receipt{ID: ID()}
|
|
now := time.Now().Unix()
|
|
result, err := tx.ExecContext(ctx, "INSERT INTO receipts(id,scope,key,digest,created) VALUES(?,?,?,?,?) ON CONFLICT(scope,key) DO NOTHING", r.ID, scope, key, digest, now)
|
|
if err != nil {
|
|
return Receipt{}, err
|
|
}
|
|
n, _ := result.RowsAffected()
|
|
if n == 0 {
|
|
var old string
|
|
if err := tx.QueryRowContext(ctx, "SELECT id,digest FROM receipts WHERE scope=? AND key=?", scope, key).Scan(&r.ID, &old); err != nil {
|
|
return Receipt{}, err
|
|
}
|
|
if old != digest {
|
|
return Receipt{}, ErrConflict
|
|
}
|
|
r.Duplicate = true
|
|
} else {
|
|
for _, j := range jobs {
|
|
if _, err := tx.ExecContext(ctx, "INSERT INTO deliveries(id,receipt_id,mapping_id,target,data,state,created,next) VALUES(?,?,?,?,?,'pending',?,?)", ID(), r.ID, j.MappingID, j.Target, []byte(j.Data), now, now); err != nil {
|
|
return Receipt{}, err
|
|
}
|
|
}
|
|
}
|
|
return r, tx.Commit()
|
|
}
|
|
|
|
func (s *Store) Claim(ctx context.Context, now time.Time) (*Job, error) {
|
|
j := new(Job)
|
|
lease := ID()
|
|
err := s.db.QueryRowContext(ctx, `UPDATE deliveries SET state='sending',attempts=attempts+1,next=?,lease=? WHERE id=(SELECT id FROM deliveries WHERE state IN ('pending','sending') AND next<=? ORDER BY next,created LIMIT 1) RETURNING id,receipt_id,mapping_id,target,data,state,attempts,created,next,status,error,lease`, now.Add(10*time.Minute).Unix(), lease, now.Unix()).Scan(&j.ID, &j.ReceiptID, &j.MappingID, &j.Target, &j.Data, &j.State, &j.Attempts, &j.Created, &j.Next, &j.Status, &j.Error, &j.Lease)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return nil, nil
|
|
}
|
|
return j, err
|
|
}
|
|
|
|
func (s *Store) Finish(ctx context.Context, j Job, state string, status int, message string, next time.Time) error {
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback()
|
|
r, err := tx.ExecContext(ctx, "UPDATE deliveries SET state=?,status=?,error=?,next=?,lease='' WHERE id=? AND lease=? AND state='sending'", state, status, message, next.Unix(), j.ID, j.Lease)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
n, _ := r.RowsAffected()
|
|
if n != 1 {
|
|
return errors.New("delivery lease lost")
|
|
}
|
|
_, err = tx.ExecContext(ctx, "INSERT INTO attempts(delivery_id,at,status,error,state) VALUES(?,?,?,?,?)", j.ID, time.Now().Unix(), status, message, state)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return tx.Commit()
|
|
}
|
|
|
|
func (s *Store) Retry(ctx context.Context, id string) error {
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback()
|
|
r, err := tx.ExecContext(ctx, "UPDATE deliveries SET state='pending',attempts=0,next=?,error='' WHERE id=? AND state='dead'", time.Now().Unix(), id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
n, _ := r.RowsAffected()
|
|
if n != 1 {
|
|
return ErrNotRetryable
|
|
}
|
|
_, err = tx.ExecContext(ctx, "INSERT INTO attempts(delivery_id,at,status,error,state) VALUES(?,?,0,'','manual_retry')", id, time.Now().Unix())
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return tx.Commit()
|
|
}
|
|
|
|
func (s *Store) List(ctx context.Context, limit, offset int) ([]Job, error) {
|
|
if limit < 1 || limit > 200 {
|
|
limit = 50
|
|
}
|
|
if offset < 0 {
|
|
offset = 0
|
|
}
|
|
rows, err := s.db.QueryContext(ctx, "SELECT id,receipt_id,mapping_id,target,state,attempts,created,next,status,error FROM deliveries ORDER BY created DESC,rowid DESC LIMIT ? OFFSET ?", limit, offset)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
out := []Job{}
|
|
for rows.Next() {
|
|
var j Job
|
|
if err := rows.Scan(&j.ID, &j.ReceiptID, &j.MappingID, &j.Target, &j.State, &j.Attempts, &j.Created, &j.Next, &j.Status, &j.Error); err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, j)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
func (s *Store) History(ctx context.Context, id string) ([]Attempt, error) {
|
|
rows, err := s.db.QueryContext(ctx, "SELECT at,status,error,state FROM attempts WHERE delivery_id=? ORDER BY id DESC LIMIT 200", id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
out := []Attempt{}
|
|
for rows.Next() {
|
|
var a Attempt
|
|
if err := rows.Scan(&a.At, &a.Status, &a.Error, &a.State); err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, a)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
func (s *Store) Counts(ctx context.Context) (map[string]int64, error) {
|
|
out := map[string]int64{"pending": 0, "sending": 0, "succeeded": 0, "dry_run": 0, "dead": 0}
|
|
rows, err := s.db.QueryContext(ctx, "SELECT state,count(*) FROM deliveries GROUP BY state")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
for rows.Next() {
|
|
var state string
|
|
var n int64
|
|
if err := rows.Scan(&state, &n); err != nil {
|
|
rows.Close()
|
|
return nil, err
|
|
}
|
|
out[state] = n
|
|
}
|
|
err = rows.Err()
|
|
rows.Close()
|
|
return out, err
|
|
}
|
|
func (s *Store) Checkpoint(ctx context.Context, id string) (string, error) {
|
|
var v string
|
|
err := s.db.QueryRowContext(ctx, "SELECT value FROM checkpoints WHERE id=?", id).Scan(&v)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return "", nil
|
|
}
|
|
return v, err
|
|
}
|
|
func (s *Store) SetCheckpoint(ctx context.Context, id, value string) error {
|
|
_, err := s.db.ExecContext(ctx, "INSERT INTO checkpoints(id,value) VALUES(?,?) ON CONFLICT(id) DO UPDATE SET value=excluded.value", id, value)
|
|
return err
|
|
}
|