| 1 | package legacycleanup |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "os" |
| 9 | "path/filepath" |
| 10 | "sort" |
| 11 | "strings" |
| 12 | "sync" |
| 13 | "time" |
| 14 | |
| 15 | "reasonix/internal/fileutil" |
| 16 | filelock "reasonix/internal/identitylock" |
| 17 | ) |
| 18 | |
| 19 | const SchemaVersion = 1 |
| 20 | |
| 21 | var ( |
| 22 | ErrNotInitialized = errors.New("legacy empty session cleanup is not initialized") |
| 23 | ErrUnsupportedVersion = errors.New("legacy empty session cleanup version is unsupported") |
| 24 | ErrCorruptState = errors.New("legacy empty session cleanup state is corrupt") |
| 25 | ) |
| 26 | |
| 27 | type TopicSnapshot struct { |
| 28 | Scope string `json:"scope"` |
| 29 | WorkspaceRoot string `json:"workspaceRoot,omitempty"` |
| 30 | TitleSource string `json:"titleSource,omitempty"` |
| 31 | CreatedAt int64 `json:"createdAt,omitempty"` |
| 32 | RowRevision int64 `json:"rowRevision,omitempty"` |
| 33 | Order int `json:"order"` |
| 34 | Pinned bool `json:"pinned,omitempty"` |
| 35 | GroupID string `json:"groupId,omitempty"` |
| 36 | GroupOrder int `json:"groupOrder,omitempty"` |
| 37 | } |
| 38 | |
| 39 | type SourceSnapshot struct { |
| 40 | Path string `json:"path"` |
| 41 | HeadID string `json:"headId,omitempty"` |
| 42 | Fingerprint string `json:"fingerprint,omitempty"` |
| 43 | } |
| 44 | |
| 45 | type Candidate struct { |
| 46 | ID string `json:"id"` |
| 47 | Kind string `json:"kind"` |
| 48 | WorkspaceID string `json:"workspaceId,omitempty"` |
| 49 | SessionID string `json:"sessionId,omitempty"` |
| 50 | TopicID string `json:"topicId,omitempty"` |
| 51 | SourcePath string `json:"sourcePath,omitempty"` |
| 52 | SourceHeadID string `json:"sourceHeadId,omitempty"` |
| 53 | SourceFingerprint string `json:"sourceFingerprint,omitempty"` |
| 54 | Sources []SourceSnapshot `json:"sources,omitempty"` |
| 55 | Title string `json:"title"` |
| 56 | TitleSequence uint64 `json:"titleSequence,omitempty"` |
| 57 | EventSequence uint64 `json:"eventSequence,omitempty"` |
| 58 | LifecycleGeneration uint64 `json:"lifecycleGeneration,omitempty"` |
| 59 | OperationID string `json:"operationId"` |
| 60 | Phase string `json:"phase"` |
| 61 | Classification string `json:"classification,omitempty"` |
| 62 | Reason string `json:"reason,omitempty"` |
| 63 | ArchivedAt int64 `json:"archivedAt,omitempty"` |
| 64 | Restored bool `json:"restored,omitempty"` |
| 65 | Topic *TopicSnapshot `json:"topic,omitempty"` |
| 66 | } |
| 67 | |
| 68 | type State struct { |
| 69 | Version int `json:"version"` |
| 70 | BatchID string `json:"batchId"` |
| 71 | RegisteredAt time.Time `json:"registeredAt"` |
| 72 | Registration string `json:"registration"` |
| 73 | Items map[string]Candidate `json:"items"` |
| 74 | } |
| 75 | |
| 76 | type Store struct { |
| 77 | path string |
| 78 | mu sync.Mutex |
| 79 | } |
| 80 | |
| 81 | func New(path string) *Store { return &Store{path: filepath.Clean(path)} } |
| 82 | |
| 83 | func (s *Store) Path() string { |
| 84 | if s == nil || s.path == "." { |
| 85 | return "" |
| 86 | } |
| 87 | return s.path |
| 88 | } |
| 89 | |
| 90 | // TryAcquireWorker gives one process exclusive ownership of the background |
| 91 | // cleanup pass. Per-update locking protects the sidecar bytes, but it cannot |
| 92 | // prevent two processes from acting on the same candidate between updates. |
| 93 | func (s *Store) TryAcquireWorker() (func(), error) { |
| 94 | if s == nil || s.Path() == "" { |
| 95 | return nil, errors.New("legacy cleanup state path is unavailable") |
| 96 | } |
| 97 | return filelock.TryAcquire(s.path + ".worker.lock") |
| 98 | } |
| 99 | |
| 100 | func (s *Store) Load(ctx context.Context) (State, error) { |
| 101 | return s.withLock(ctx, false, func() (State, error) { return load(s.path) }) |
| 102 | } |
| 103 | |
| 104 | func (s *Store) Initialize(ctx context.Context, state State) (State, bool, error) { |
| 105 | var created bool |
| 106 | result, err := s.withLock(ctx, true, func() (State, error) { |
| 107 | current, err := load(s.path) |
| 108 | if err == nil { |
| 109 | return current, nil |
| 110 | } |
| 111 | if !errors.Is(err, ErrNotInitialized) { |
| 112 | return State{}, err |
| 113 | } |
| 114 | state.Version = SchemaVersion |
| 115 | state.Registration = "complete" |
| 116 | if state.RegisteredAt.IsZero() { |
| 117 | state.RegisteredAt = time.Now().UTC() |
| 118 | } |
| 119 | if strings.TrimSpace(state.BatchID) == "" || state.Items == nil { |
| 120 | return State{}, fmt.Errorf("%w: incomplete initial state", ErrCorruptState) |
| 121 | } |
| 122 | if err := save(s.path, state); err != nil { |
| 123 | return State{}, err |
| 124 | } |
| 125 | created = true |
| 126 | return clone(state) |
| 127 | }) |
| 128 | return result, created, err |
| 129 | } |
| 130 | |
| 131 | func (s *Store) Update(ctx context.Context, mutate func(*State) error) (State, error) { |
| 132 | return s.withLock(ctx, true, func() (State, error) { |
| 133 | state, err := load(s.path) |
| 134 | if err != nil { |
| 135 | return State{}, err |
| 136 | } |
| 137 | if mutate != nil { |
| 138 | if err := mutate(&state); err != nil { |
| 139 | return State{}, err |
| 140 | } |
| 141 | } |
| 142 | if err := validate(state); err != nil { |
| 143 | return State{}, err |
| 144 | } |
| 145 | if err := save(s.path, state); err != nil { |
| 146 | return State{}, err |
| 147 | } |
| 148 | return clone(state) |
| 149 | }) |
| 150 | } |
| 151 | |
| 152 | // Transition persists a prepare state, runs one external effect while the |
| 153 | // cross-process state lock remains held, then persists the effect outcome. |
| 154 | // The callbacks must not call this Store. Persisting prepare before effect |
| 155 | // makes an interrupted archive distinguishable from one never attempted. |
| 156 | func (s *Store) Transition(ctx context.Context, prepare func(*State) error, effect func() error, finish func(*State, error) error) (State, error) { |
| 157 | return s.withLock(ctx, true, func() (State, error) { |
| 158 | state, err := load(s.path) |
| 159 | if err != nil { |
| 160 | return State{}, err |
| 161 | } |
| 162 | if prepare != nil { |
| 163 | if err := prepare(&state); err != nil { |
| 164 | return State{}, err |
| 165 | } |
| 166 | } |
| 167 | if err := validate(state); err != nil { |
| 168 | return State{}, err |
| 169 | } |
| 170 | if err := save(s.path, state); err != nil { |
| 171 | return State{}, err |
| 172 | } |
| 173 | effectErr := error(nil) |
| 174 | if effect != nil { |
| 175 | effectErr = effect() |
| 176 | } |
| 177 | if finish != nil { |
| 178 | if err := finish(&state, effectErr); err != nil { |
| 179 | return State{}, err |
| 180 | } |
| 181 | } |
| 182 | if err := validate(state); err != nil { |
| 183 | return State{}, err |
| 184 | } |
| 185 | if err := save(s.path, state); err != nil { |
| 186 | return State{}, err |
| 187 | } |
| 188 | result, err := clone(state) |
| 189 | if err != nil { |
| 190 | return State{}, err |
| 191 | } |
| 192 | return result, effectErr |
| 193 | }) |
| 194 | } |
| 195 | |
| 196 | func (s *Store) withLock(ctx context.Context, _ bool, fn func() (State, error)) (State, error) { |
| 197 | if s == nil || s.Path() == "" { |
| 198 | return State{}, errors.New("legacy cleanup state path is unavailable") |
| 199 | } |
| 200 | s.mu.Lock() |
| 201 | defer s.mu.Unlock() |
| 202 | // The advisory lock lives beside the state file, so even a first read needs |
| 203 | // the parent directory. Creating only the directory and lock file does not |
| 204 | // initialize or overwrite the versioned state. |
| 205 | if err := os.MkdirAll(filepath.Dir(s.path), 0o700); err != nil { |
| 206 | return State{}, err |
| 207 | } |
| 208 | release, err := filelock.Acquire(ctx, s.path+".lock") |
| 209 | if err != nil { |
| 210 | return State{}, err |
| 211 | } |
| 212 | defer release() |
| 213 | return fn() |
| 214 | } |
| 215 | |
| 216 | func load(path string) (State, error) { |
| 217 | body, err := os.ReadFile(path) |
| 218 | if os.IsNotExist(err) { |
| 219 | return State{}, ErrNotInitialized |
| 220 | } |
| 221 | if err != nil { |
| 222 | return State{}, err |
| 223 | } |
| 224 | var header struct { |
| 225 | Version int `json:"version"` |
| 226 | } |
| 227 | if err := json.Unmarshal(body, &header); err != nil { |
| 228 | return State{}, fmt.Errorf("%w: %w", ErrCorruptState, err) |
| 229 | } |
| 230 | if header.Version != SchemaVersion { |
| 231 | return State{}, fmt.Errorf("%w: %d", ErrUnsupportedVersion, header.Version) |
| 232 | } |
| 233 | var state State |
| 234 | if err := json.Unmarshal(body, &state); err != nil { |
| 235 | return State{}, fmt.Errorf("%w: %w", ErrCorruptState, err) |
| 236 | } |
| 237 | if err := validate(state); err != nil { |
| 238 | return State{}, err |
| 239 | } |
| 240 | return state, nil |
| 241 | } |
| 242 | |
| 243 | func validate(state State) error { |
| 244 | if state.Version != SchemaVersion || strings.TrimSpace(state.BatchID) == "" || state.Registration != "complete" || state.Items == nil { |
| 245 | return fmt.Errorf("%w: invalid header", ErrCorruptState) |
| 246 | } |
| 247 | for id, item := range state.Items { |
| 248 | if id == "" || item.ID != id || (item.Kind != "session" && item.Kind != "legacy" && item.Kind != "topic") || item.OperationID == "" || item.Phase == "" { |
| 249 | return fmt.Errorf("%w: invalid candidate %q", ErrCorruptState, id) |
| 250 | } |
| 251 | if item.Kind == "legacy" && strings.TrimSpace(item.SourcePath) == "" { |
| 252 | return fmt.Errorf("%w: legacy candidate %q has no source", ErrCorruptState, id) |
| 253 | } |
| 254 | for _, source := range item.Sources { |
| 255 | if strings.TrimSpace(source.Path) == "" { |
| 256 | return fmt.Errorf("%w: candidate %q has an invalid source", ErrCorruptState, id) |
| 257 | } |
| 258 | } |
| 259 | } |
| 260 | return nil |
| 261 | } |
| 262 | |
| 263 | func save(path string, state State) error { |
| 264 | body, err := json.MarshalIndent(state, "", " ") |
| 265 | if err != nil { |
| 266 | return err |
| 267 | } |
| 268 | body = append(body, '\n') |
| 269 | return fileutil.AtomicWriteFileStrict(path, body, 0o600) |
| 270 | } |
| 271 | |
| 272 | func clone(state State) (State, error) { |
| 273 | body, err := json.Marshal(state) |
| 274 | if err != nil { |
| 275 | return State{}, err |
| 276 | } |
| 277 | var result State |
| 278 | if err := json.Unmarshal(body, &result); err != nil { |
| 279 | return State{}, err |
| 280 | } |
| 281 | return result, nil |
| 282 | } |
| 283 | |
| 284 | func SortedItems(state State) []Candidate { |
| 285 | items := make([]Candidate, 0, len(state.Items)) |
| 286 | for _, item := range state.Items { |
| 287 | items = append(items, item) |
| 288 | } |
| 289 | sort.Slice(items, func(i, j int) bool { return items[i].ID < items[j].ID }) |
| 290 | return items |
| 291 | } |
| 292 |