| 1 | package moves |
| 2 | |
| 3 | import ( |
| 4 | "crypto/sha256" |
| 5 | "encoding/hex" |
| 6 | "encoding/json" |
| 7 | "fmt" |
| 8 | "sort" |
| 9 | "time" |
| 10 | |
| 11 | "github.com/carlosframework/platform/internal/registry" |
| 12 | ) |
| 13 | |
| 14 | // timerWire is registry.Timer with the registration on the wire: the |
| 15 | // snapshot carries it so ReplaceTimers can preserve it (§9.4). |
| 16 | type timerWire struct { |
| 17 | Host, Name, Kind, Spec, Path string |
| 18 | NextAt, LastAt int64 |
| 19 | LastStatus string |
| 20 | LastMs int64 |
| 21 | Attempts int |
| 22 | CreatedAt, DueAt int64 |
| 23 | Registration string |
| 24 | } |
| 25 | |
| 26 | func toWire(t registry.Timer) timerWire { |
| 27 | return timerWire{t.Host, t.Name, t.Kind, t.Spec, t.Path, t.NextAt, t.LastAt, t.LastStatus, t.LastMs, t.Attempts, t.CreatedAt, t.DueAt, t.Registration} |
| 28 | } |
| 29 | |
| 30 | func fromWire(w timerWire) registry.Timer { |
| 31 | return registry.Timer{Host: w.Host, Name: w.Name, Kind: w.Kind, Spec: w.Spec, Path: w.Path, NextAt: w.NextAt, LastAt: w.LastAt, LastStatus: w.LastStatus, LastMs: w.LastMs, Attempts: w.Attempts, CreatedAt: w.CreatedAt, DueAt: w.DueAt, Registration: w.Registration} |
| 32 | } |
| 33 | |
| 34 | // TimersSnapshot is the timers table for one host as carried (§6.2, §9.4). |
| 35 | type TimersSnapshot struct { |
| 36 | V int `json:"v"` |
| 37 | Host string `json:"host"` |
| 38 | Rows []registry.Timer `json:"-"` |
| 39 | TakenAt time.Time `json:"taken_at"` |
| 40 | } |
| 41 | |
| 42 | type timersWire struct { |
| 43 | V int `json:"v"` |
| 44 | Host string `json:"host"` |
| 45 | Rows []timerWire `json:"rows"` |
| 46 | TakenAt time.Time `json:"taken_at"` |
| 47 | } |
| 48 | |
| 49 | func (s TimersSnapshot) Encode() ([]byte, error) { |
| 50 | w := timersWire{V: timersVersion, Host: s.Host, TakenAt: s.TakenAt} |
| 51 | for _, r := range s.Rows { |
| 52 | w.Rows = append(w.Rows, toWire(r)) |
| 53 | } |
| 54 | return json.Marshal(w) |
| 55 | } |
| 56 | |
| 57 | // timersVersion is the only snapshot schema this reader understands. |
| 58 | const timersVersion = 1 |
| 59 | |
| 60 | // DecodeTimers reads a v1 snapshot. Any other version, a missing version |
| 61 | // and a JSON null are refused before any row is returned: a newer schema |
| 62 | // could carry state this reader would drop, and re-encoding would then |
| 63 | // label the partial reading v1. |
| 64 | func DecodeTimers(b []byte) (TimersSnapshot, error) { |
| 65 | var w timersWire |
| 66 | if err := json.Unmarshal(b, &w); err != nil { |
| 67 | return TimersSnapshot{}, fmt.Errorf("moves: timers snapshot: %w", err) |
| 68 | } |
| 69 | if w.V != timersVersion { |
| 70 | return TimersSnapshot{}, fmt.Errorf("moves: timers snapshot: unsupported version %d (want %d)", w.V, timersVersion) |
| 71 | } |
| 72 | s := TimersSnapshot{V: w.V, Host: w.Host, TakenAt: w.TakenAt} |
| 73 | for _, r := range w.Rows { |
| 74 | s.Rows = append(s.Rows, fromWire(r)) |
| 75 | } |
| 76 | return s, nil |
| 77 | } |
| 78 | |
| 79 | // ActiveTimers drops disposable history: a once-timer that has fired and |
| 80 | // has no next occurrence. Everything else is acknowledged state (INV-2). |
| 81 | func ActiveTimers(rows []registry.Timer) []registry.Timer { |
| 82 | var out []registry.Timer |
| 83 | for _, t := range rows { |
| 84 | if t.Kind == "once" && t.NextAt == 0 && t.LastStatus != "" { |
| 85 | continue |
| 86 | } |
| 87 | out = append(out, t) |
| 88 | } |
| 89 | return out |
| 90 | } |
| 91 | |
| 92 | // TimersDigest is the canonical digest a clean recovery compares (§9.4): |
| 93 | // active rows, sorted by name then kind, every bookkeeping column. |
| 94 | func TimersDigest(rows []registry.Timer) string { |
| 95 | active := ActiveTimers(rows) |
| 96 | sort.Slice(active, func(i, j int) bool { |
| 97 | if active[i].Name != active[j].Name { |
| 98 | return active[i].Name < active[j].Name |
| 99 | } |
| 100 | return active[i].Kind < active[j].Kind |
| 101 | }) |
| 102 | h := sha256.New() |
| 103 | for _, t := range active { |
| 104 | b, _ := json.Marshal(toWire(t)) |
| 105 | h.Write(b) |
| 106 | h.Write([]byte{'\n'}) |
| 107 | } |
| 108 | return "sha256:" + hex.EncodeToString(h.Sum(nil)) |
| 109 | } |
| 110 | |
amadan