278 lines
10 KiB
Go
278 lines
10 KiB
Go
package store
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/wotra/wotra/internal/domain"
|
|
)
|
|
|
|
// ErrSyncStale is returned when an existing client is behind the prune marker.
|
|
var ErrSyncStale = errors.New("sync state stale: full re-sync required")
|
|
|
|
const pruneEntity = "_pruned"
|
|
const pruneOp = "marker"
|
|
|
|
// SyncStore manages the sync_log.
|
|
type SyncStore struct {
|
|
db *sql.DB
|
|
}
|
|
|
|
func NewSyncStore(db *sql.DB) *SyncStore {
|
|
return &SyncStore{db: db}
|
|
}
|
|
|
|
type SyncChange struct {
|
|
Entity string `json:"entity"`
|
|
EntityID string `json:"entity_id"`
|
|
Op string `json:"op"`
|
|
Version int64 `json:"version"`
|
|
Payload string `json:"payload"`
|
|
}
|
|
|
|
// Bootstrap returns the current contents of every client-replicated entity.
|
|
// It deliberately bypasses sync_log so retention cannot hide old data.
|
|
func (s *SyncStore) Bootstrap(ctx context.Context) ([]SyncChange, int64, error) {
|
|
var changes []SyncChange
|
|
add := func(entity, id string, value any) error {
|
|
payload, err := json.Marshal(value)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
changes = append(changes, SyncChange{Entity: entity, EntityID: id, Op: "upsert", Payload: string(payload)})
|
|
return nil
|
|
}
|
|
|
|
rows, err := s.db.QueryContext(ctx, `SELECT id, start_time, end_time, auto_stopped, note, day_key, updated_at, deleted_at FROM entries WHERE deleted_at IS NULL`)
|
|
if err != nil { return nil, 0, err }
|
|
for rows.Next() {
|
|
var e domain.Entry; var end, deleted sql.NullInt64; var stopped int
|
|
if err := rows.Scan(&e.ID, &e.StartTime, &end, &stopped, &e.Note, &e.DayKey, &e.UpdatedAt, &deleted); err != nil { rows.Close(); return nil, 0, err }
|
|
if end.Valid { e.EndTime = &end.Int64 }; e.AutoStopped = stopped != 0
|
|
if err := add("entries", e.ID, e); err != nil { rows.Close(); return nil, 0, err }
|
|
}
|
|
if err := rows.Err(); err != nil { rows.Close(); return nil, 0, err }; rows.Close()
|
|
|
|
rows, err = s.db.QueryContext(ctx, `SELECT day_key, start_time, end_time, worked_ms, kind, closed_at, updated_at FROM closed_days`)
|
|
if err != nil { return nil, 0, err }
|
|
for rows.Next() {
|
|
var d domain.ClosedDay; var start, end sql.NullInt64
|
|
if err := rows.Scan(&d.DayKey, &start, &end, &d.WorkedMs, &d.Kind, &d.ClosedAt, &d.UpdatedAt); err != nil { rows.Close(); return nil, 0, err }
|
|
if start.Valid { d.StartTime = &start.Int64 }; if end.Valid { d.EndTime = &end.Int64 }
|
|
if err := add("closed_days", d.DayKey, d); err != nil { rows.Close(); return nil, 0, err }
|
|
}
|
|
if err := rows.Err(); err != nil { rows.Close(); return nil, 0, err }; rows.Close()
|
|
|
|
rows, err = s.db.QueryContext(ctx, `SELECT week_key, expected_ms, worked_ms, delta_ms, closed_at, updated_at FROM closed_weeks`)
|
|
if err != nil { return nil, 0, err }
|
|
for rows.Next() {
|
|
var w domain.ClosedWeek
|
|
if err := rows.Scan(&w.WeekKey, &w.ExpectedMs, &w.WorkedMs, &w.DeltaMs, &w.ClosedAt, &w.UpdatedAt); err != nil { rows.Close(); return nil, 0, err }
|
|
if err := add("closed_weeks", w.WeekKey, w); err != nil { rows.Close(); return nil, 0, err }
|
|
}
|
|
if err := rows.Err(); err != nil { rows.Close(); return nil, 0, err }; rows.Close()
|
|
|
|
rows, err = s.db.QueryContext(ctx, `SELECT id, effective_from, hours_per_week, workdays_mask, timezone, created_at, updated_at FROM settings_history`)
|
|
if err != nil { return nil, 0, err }
|
|
for rows.Next() {
|
|
var set domain.Settings
|
|
if err := rows.Scan(&set.ID, &set.EffectiveFrom, &set.HoursPerWeek, &set.WorkdaysMask, &set.Timezone, &set.CreatedAt, &set.UpdatedAt); err != nil { rows.Close(); return nil, 0, err }
|
|
if err := add("settings_history", set.ID, set); err != nil { rows.Close(); return nil, 0, err }
|
|
}
|
|
if err := rows.Err(); err != nil { rows.Close(); return nil, 0, err }; rows.Close()
|
|
|
|
rows, err = s.db.QueryContext(ctx, `SELECT id, delta_ms, note, effective_at, created_at, updated_at FROM balance_adjustments`)
|
|
if err != nil { return nil, 0, err }
|
|
for rows.Next() {
|
|
var a domain.BalanceAdjustment
|
|
if err := rows.Scan(&a.ID, &a.DeltaMs, &a.Note, &a.EffectiveAt, &a.CreatedAt, &a.UpdatedAt); err != nil { rows.Close(); return nil, 0, err }
|
|
if err := add("balance_adjustments", a.ID, a); err != nil { rows.Close(); return nil, 0, err }
|
|
}
|
|
if err := rows.Err(); err != nil { rows.Close(); return nil, 0, err }; rows.Close()
|
|
|
|
var version int64
|
|
if err := s.db.QueryRowContext(ctx, `SELECT COALESCE(MAX(version), 0) FROM sync_log`).Scan(&version); err != nil { return nil, 0, err }
|
|
return changes, version, nil
|
|
}
|
|
|
|
// Pull returns all sync_log rows with version > sinceVersion.
|
|
// It calls Prune first with a 30-day TTL. Clients behind a prune marker
|
|
// must bootstrap from current entities before resuming incremental sync.
|
|
func (s *SyncStore) Pull(ctx context.Context, sinceVersion int64) ([]SyncChange, int64, error) {
|
|
if err := s.Prune(ctx, 30*24*time.Hour); err != nil {
|
|
return nil, 0, err
|
|
}
|
|
|
|
rows, err := s.db.QueryContext(ctx,
|
|
`SELECT entity, entity_id, op, version, payload FROM sync_log
|
|
WHERE version > ? ORDER BY version ASC`, sinceVersion)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var changes []SyncChange
|
|
var maxVersion int64 = sinceVersion
|
|
for rows.Next() {
|
|
var c SyncChange
|
|
if err := rows.Scan(&c.Entity, &c.EntityID, &c.Op, &c.Version, &c.Payload); err != nil {
|
|
return nil, 0, err
|
|
}
|
|
if c.Entity == pruneEntity {
|
|
return nil, 0, ErrSyncStale
|
|
}
|
|
if c.Version > maxVersion {
|
|
maxVersion = c.Version
|
|
}
|
|
changes = append(changes, c)
|
|
}
|
|
return changes, maxVersion, rows.Err()
|
|
}
|
|
|
|
// Prune deletes sync_log rows older than ttl and inserts a prune marker at the
|
|
// version boundary so stale clients can detect they need a full re-sync.
|
|
func (s *SyncStore) Prune(ctx context.Context, ttl time.Duration) error {
|
|
cutoff := time.Now().Add(-ttl).UnixMilli()
|
|
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback() //nolint:errcheck
|
|
|
|
// Find max version among rows that will be pruned (excluding existing markers).
|
|
var maxPruned sql.NullInt64
|
|
err = tx.QueryRowContext(ctx,
|
|
`SELECT MAX(version) FROM sync_log WHERE logged_at < ? AND entity != ?`,
|
|
cutoff, pruneEntity).Scan(&maxPruned)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !maxPruned.Valid {
|
|
// Nothing to prune.
|
|
return tx.Commit()
|
|
}
|
|
|
|
// Delete old rows (but not the existing marker, if any).
|
|
if _, err = tx.ExecContext(ctx,
|
|
`DELETE FROM sync_log WHERE logged_at < ? AND entity != ?`,
|
|
cutoff, pruneEntity); err != nil {
|
|
return err
|
|
}
|
|
|
|
// Insert (or replace) the prune marker at the boundary version.
|
|
now := time.Now().UnixMilli()
|
|
if _, err = tx.ExecContext(ctx,
|
|
`INSERT OR REPLACE INTO sync_log (entity, entity_id, op, version, payload, logged_at)
|
|
VALUES (?, ?, ?, ?, '{}', ?)`,
|
|
pruneEntity, pruneEntity, pruneOp, maxPruned.Int64, now); err != nil {
|
|
return err
|
|
}
|
|
|
|
return tx.Commit()
|
|
}
|
|
|
|
// nextVersion returns the next monotonic version number.
|
|
func (s *SyncStore) nextVersion(ctx context.Context) (int64, error) {
|
|
var max sql.NullInt64
|
|
err := s.db.QueryRowContext(ctx, `SELECT MAX(version) FROM sync_log`).Scan(&max)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
if !max.Valid {
|
|
return 1, nil
|
|
}
|
|
return max.Int64 + 1, nil
|
|
}
|
|
|
|
// LogEntry appends an entry upsert to the sync log.
|
|
func (s *SyncStore) LogEntry(ctx context.Context, e *domain.Entry) error {
|
|
payload, err := json.Marshal(e)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return s.log(ctx, "entries", e.ID, "upsert", string(payload))
|
|
}
|
|
|
|
// LogEntryDelete appends an entry delete to the sync log.
|
|
func (s *SyncStore) LogEntryDelete(ctx context.Context, id string) error {
|
|
payload := fmt.Sprintf(`{"id":%q}`, id)
|
|
return s.log(ctx, "entries", id, "delete", payload)
|
|
}
|
|
|
|
// LogClosedDay appends a closed_day upsert to the sync log.
|
|
func (s *SyncStore) LogClosedDay(ctx context.Context, d *domain.ClosedDay) error {
|
|
payload, err := json.Marshal(d)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return s.log(ctx, "closed_days", d.DayKey, "upsert", string(payload))
|
|
}
|
|
|
|
// LogClosedDayDelete appends a closed_day delete to the sync log.
|
|
func (s *SyncStore) LogClosedDayDelete(ctx context.Context, dayKey string) error {
|
|
payload := fmt.Sprintf(`{"day_key":%q}`, dayKey)
|
|
return s.log(ctx, "closed_days", dayKey, "delete", payload)
|
|
}
|
|
|
|
// LogClosedWeek appends a closed_week upsert to the sync log.
|
|
func (s *SyncStore) LogClosedWeek(ctx context.Context, w *domain.ClosedWeek) error {
|
|
payload, err := json.Marshal(w)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return s.log(ctx, "closed_weeks", w.WeekKey, "upsert", string(payload))
|
|
}
|
|
|
|
// LogClosedWeekDelete appends a closed_week delete to the sync log.
|
|
func (s *SyncStore) LogClosedWeekDelete(ctx context.Context, weekKey string) error {
|
|
payload := fmt.Sprintf(`{"week_key":%q}`, weekKey)
|
|
return s.log(ctx, "closed_weeks", weekKey, "delete", payload)
|
|
}
|
|
|
|
// LogSettings appends a settings upsert to the sync log.
|
|
func (s *SyncStore) LogSettings(ctx context.Context, set *domain.Settings) error {
|
|
payload, err := json.Marshal(set)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return s.log(ctx, "settings_history", set.ID, "upsert", string(payload))
|
|
}
|
|
|
|
// LogSettingsDelete appends a settings delete to the sync log.
|
|
func (s *SyncStore) LogSettingsDelete(ctx context.Context, id string) error {
|
|
payload := fmt.Sprintf(`{"id":%q}`, id)
|
|
return s.log(ctx, "settings_history", id, "delete", payload)
|
|
}
|
|
|
|
// LogBalanceAdjustment appends a balance_adjustment upsert to the sync log.
|
|
func (s *SyncStore) LogBalanceAdjustment(ctx context.Context, a *domain.BalanceAdjustment) error {
|
|
payload, err := json.Marshal(a)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return s.log(ctx, "balance_adjustments", a.ID, "upsert", string(payload))
|
|
}
|
|
|
|
// LogBalanceAdjustmentDelete appends a balance_adjustment delete to the sync log.
|
|
func (s *SyncStore) LogBalanceAdjustmentDelete(ctx context.Context, id string) error {
|
|
payload := fmt.Sprintf(`{"id":%q}`, id)
|
|
return s.log(ctx, "balance_adjustments", id, "delete", payload)
|
|
}
|
|
|
|
func (s *SyncStore) log(ctx context.Context, entity, entityID, op, payload string) error {
|
|
version, err := s.nextVersion(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
now := time.Now().UnixMilli()
|
|
_, err = s.db.ExecContext(ctx,
|
|
`INSERT INTO sync_log (entity, entity_id, op, version, payload, logged_at) VALUES (?, ?, ?, ?, ?, ?)`,
|
|
entity, entityID, op, version, payload, now)
|
|
return err
|
|
}
|