| 1 | // Package taskcatalog maintains a disposable cross-project projection of the |
| 2 | // authoritative taskmonitor FileStore. |
| 3 | package taskcatalog |
| 4 | |
| 5 | import ( |
| 6 | "context" |
| 7 | "crypto/sha256" |
| 8 | "database/sql" |
| 9 | "encoding/base64" |
| 10 | "encoding/hex" |
| 11 | "encoding/json" |
| 12 | "errors" |
| 13 | "fmt" |
| 14 | "os" |
| 15 | "path/filepath" |
| 16 | "strings" |
| 17 | "sync" |
| 18 | "sync/atomic" |
| 19 | "time" |
| 20 | |
| 21 | "reasonix/internal/config" |
| 22 | "reasonix/internal/projectiondb" |
| 23 | "reasonix/internal/taskmonitor" |
| 24 | ) |
| 25 | |
| 26 | const ( |
| 27 | SchemaVersion = 1 |
| 28 | DefaultLimit = 50 |
| 29 | MaxLimit = 200 |
| 30 | missingGrace = 30 * time.Second |
| 31 | ) |
| 32 | |
| 33 | type Status struct { |
| 34 | State string `json:"state"` |
| 35 | Mode projectiondb.Mode `json:"mode"` |
| 36 | Path string `json:"path,omitempty"` |
| 37 | Revision uint64 `json:"revision"` |
| 38 | Indexed int64 `json:"indexed"` |
| 39 | Total int64 `json:"total"` |
| 40 | Pending int64 `json:"pending"` |
| 41 | Failed int64 `json:"failed"` |
| 42 | LastError string `json:"lastError,omitempty"` |
| 43 | } |
| 44 | |
| 45 | type Project struct { |
| 46 | Key string `json:"projectKey"` |
| 47 | Root string `json:"projectRoot"` |
| 48 | Label string `json:"projectLabel"` |
| 49 | } |
| 50 | |
| 51 | type PageRequest struct { |
| 52 | ProjectKeys []string |
| 53 | SessionID string |
| 54 | States []string |
| 55 | Query string |
| 56 | Cursor string |
| 57 | Limit int |
| 58 | } |
| 59 | |
| 60 | type Item struct { |
| 61 | ProjectKey string `json:"projectKey"` |
| 62 | ProjectLabel string `json:"projectLabel"` |
| 63 | Task taskmonitor.TaskSnapshot `json:"task"` |
| 64 | } |
| 65 | |
| 66 | type Page struct { |
| 67 | Items []Item `json:"items"` |
| 68 | NextCursor string `json:"nextCursor"` |
| 69 | Revision uint64 `json:"revision"` |
| 70 | Partial bool `json:"partial"` |
| 71 | StaleCursor bool `json:"staleCursor"` |
| 72 | Status Status `json:"status"` |
| 73 | } |
| 74 | |
| 75 | type EventPage struct { |
| 76 | Items []taskmonitor.TaskEvent `json:"items"` |
| 77 | NextSequence int `json:"nextSequence"` |
| 78 | Partial bool `json:"partial"` |
| 79 | } |
| 80 | |
| 81 | type cursor struct { |
| 82 | Revision uint64 `json:"r"` |
| 83 | Updated int64 `json:"u"` |
| 84 | Project string `json:"p"` |
| 85 | Task string `json:"t"` |
| 86 | } |
| 87 | |
| 88 | type request struct { |
| 89 | projectRoot string |
| 90 | taskID string |
| 91 | events bool |
| 92 | flush chan struct{} |
| 93 | } |
| 94 | |
| 95 | type Catalog struct { |
| 96 | db *sql.DB |
| 97 | store *taskmonitor.FileStore |
| 98 | ctx context.Context |
| 99 | cancel context.CancelFunc |
| 100 | queue chan request |
| 101 | dirtyWake chan struct{} |
| 102 | dirtyProjects sync.Map |
| 103 | wg sync.WaitGroup |
| 104 | closing atomic.Bool |
| 105 | revision atomic.Uint64 |
| 106 | statusMu sync.RWMutex |
| 107 | status Status |
| 108 | projectLocks sync.Map |
| 109 | reconcileMu sync.Mutex |
| 110 | reconciling map[string]bool |
| 111 | registered map[string]bool |
| 112 | reconcileDone bool |
| 113 | closeOnce sync.Once |
| 114 | closeDone chan struct{} |
| 115 | closeErr error |
| 116 | } |
| 117 | |
| 118 | // DefaultPath returns the disposable task projection path under CacheDir. |
| 119 | // Empty when cache is unavailable so Open falls back to an in-memory projection. |
| 120 | func DefaultPath() string { |
| 121 | cache := strings.TrimSpace(config.CacheDir()) |
| 122 | if cache == "" { |
| 123 | return "" |
| 124 | } |
| 125 | return filepath.Join(cache, "task-catalog", "v1.sqlite") |
| 126 | } |
| 127 | |
| 128 | func ProjectKey(root string) string { |
| 129 | root = filepath.Clean(strings.TrimSpace(root)) |
| 130 | if abs, err := filepath.Abs(root); err == nil { |
| 131 | root = abs |
| 132 | } |
| 133 | sum := sha256.Sum256([]byte(root)) |
| 134 | return hex.EncodeToString(sum[:]) |
| 135 | } |
| 136 | |
| 137 | const schema = ` |
| 138 | CREATE TABLE task_state(id INTEGER PRIMARY KEY CHECK(id=1),revision INTEGER NOT NULL DEFAULT 0); |
| 139 | INSERT INTO task_state(id,revision) VALUES(1,0); |
| 140 | CREATE TABLE task_projects(project_key TEXT PRIMARY KEY,project_root TEXT UNIQUE NOT NULL,project_label TEXT NOT NULL DEFAULT '', |
| 141 | signature TEXT NOT NULL DEFAULT '',scan_generation INTEGER NOT NULL DEFAULT 0,scan_cursor TEXT NOT NULL DEFAULT '',state TEXT NOT NULL DEFAULT 'pending', |
| 142 | error TEXT NOT NULL DEFAULT '',indexed INTEGER NOT NULL DEFAULT 0,total INTEGER NOT NULL DEFAULT 0,completed_at INTEGER NOT NULL DEFAULT 0); |
| 143 | CREATE TABLE task_snapshots(project_key TEXT NOT NULL,task_id TEXT NOT NULL,session_id TEXT NOT NULL DEFAULT '',job_id TEXT NOT NULL DEFAULT '', |
| 144 | kind TEXT NOT NULL DEFAULT '',label TEXT NOT NULL DEFAULT '',state TEXT NOT NULL,runtime_state TEXT NOT NULL DEFAULT '',runtime_lease_until INTEGER NOT NULL DEFAULT 0, |
| 145 | version INTEGER NOT NULL,created_at INTEGER NOT NULL,updated_at INTEGER NOT NULL,error_code TEXT NOT NULL DEFAULT '',snapshot_fingerprint TEXT NOT NULL DEFAULT '', |
| 146 | snapshot_json BLOB NOT NULL,health TEXT NOT NULL DEFAULT 'ok',missing_since INTEGER NOT NULL DEFAULT 0,seen_generation INTEGER NOT NULL DEFAULT 0, |
| 147 | PRIMARY KEY(project_key,task_id),FOREIGN KEY(project_key) REFERENCES task_projects(project_key) ON DELETE CASCADE); |
| 148 | CREATE TABLE task_event_sources(project_key TEXT NOT NULL,task_id TEXT NOT NULL,path TEXT NOT NULL,size INTEGER NOT NULL DEFAULT 0,indexed_offset INTEGER NOT NULL DEFAULT 0, |
| 149 | fingerprint TEXT NOT NULL DEFAULT '',state TEXT NOT NULL DEFAULT 'pending',error TEXT NOT NULL DEFAULT '',PRIMARY KEY(project_key,task_id)); |
| 150 | CREATE TABLE task_events(project_key TEXT NOT NULL,task_id TEXT NOT NULL,sequence INTEGER NOT NULL,timestamp INTEGER NOT NULL,event_type TEXT NOT NULL, |
| 151 | state TEXT NOT NULL,runtime_state TEXT NOT NULL DEFAULT '',event_json BLOB NOT NULL,PRIMARY KEY(project_key,task_id,sequence)); |
| 152 | CREATE INDEX idx_task_project_page ON task_snapshots(project_key,updated_at DESC,task_id); |
| 153 | CREATE INDEX idx_task_state_page ON task_snapshots(state,updated_at DESC,project_key,task_id); |
| 154 | CREATE INDEX idx_task_session ON task_snapshots(session_id,updated_at DESC,task_id); |
| 155 | CREATE INDEX idx_task_events_page ON task_events(project_key,task_id,sequence); |
| 156 | ` |
| 157 | |
| 158 | func migrations() []projectiondb.Migration { |
| 159 | return []projectiondb.Migration{{Version: 1, Apply: func(ctx context.Context, tx *sql.Tx) error { |
| 160 | _, err := tx.ExecContext(ctx, schema) |
| 161 | return err |
| 162 | }}} |
| 163 | } |
| 164 | |
| 165 | func Open(ctx context.Context, path string) (*Catalog, error) { |
| 166 | if path == "" { |
| 167 | path = DefaultPath() |
| 168 | } |
| 169 | inMemory := strings.TrimSpace(path) == "" |
| 170 | if inMemory { |
| 171 | path = "" |
| 172 | } |
| 173 | handle, err := projectiondb.Open(ctx, projectiondb.OpenOptions{ |
| 174 | Path: path, MemoryName: "task-catalog", Migrations: migrations(), InMemory: inMemory, MaxOpenConns: 4, |
| 175 | }) |
| 176 | if err != nil { |
| 177 | return nil, err |
| 178 | } |
| 179 | workerCtx, cancel := context.WithCancel(context.Background()) |
| 180 | c := &Catalog{db: handle.DB, store: taskmonitor.NewFileStore(filepath.Join(".reasonix", "tasks")), ctx: workerCtx, cancel: cancel, |
| 181 | queue: make(chan request, 1024), dirtyWake: make(chan struct{}, 1), reconciling: map[string]bool{}, registered: map[string]bool{}, closeDone: make(chan struct{}), |
| 182 | status: Status{State: string(handle.Status.State), Mode: handle.Status.Mode, Path: handle.Status.Path, LastError: handle.Status.LastError}} |
| 183 | var revision uint64 |
| 184 | _ = c.db.QueryRowContext(ctx, `SELECT revision FROM task_state WHERE id=1`).Scan(&revision) |
| 185 | c.revision.Store(revision) |
| 186 | c.refresh(ctx) |
| 187 | c.wg.Add(1) |
| 188 | go c.worker() |
| 189 | return c, nil |
| 190 | } |
| 191 | |
| 192 | func (c *Catalog) ObservedStore() *taskmonitor.FileStore { |
| 193 | return taskmonitor.NewObservedFileStore(filepath.Join(".reasonix", "tasks"), c) |
| 194 | } |
| 195 | |
| 196 | func (c *Catalog) SnapshotChanged(projectRoot, taskID string) { |
| 197 | c.enqueue(request{projectRoot: projectRoot, taskID: taskID}) |
| 198 | } |
| 199 | func (c *Catalog) EventsChanged(projectRoot, taskID string) { |
| 200 | c.enqueue(request{projectRoot: projectRoot, taskID: taskID, events: true}) |
| 201 | } |
| 202 | |
| 203 | func (c *Catalog) enqueue(req request) { |
| 204 | if c.closing.Load() { |
| 205 | return |
| 206 | } |
| 207 | select { |
| 208 | case c.queue <- req: |
| 209 | default: |
| 210 | c.dirtyProjects.Store(req.projectRoot, true) |
| 211 | c.wakeDirty() |
| 212 | } |
| 213 | } |
| 214 | |
| 215 | func (c *Catalog) worker() { |
| 216 | defer c.wg.Done() |
| 217 | ticker := time.NewTicker(time.Minute) |
| 218 | defer ticker.Stop() |
| 219 | for { |
| 220 | if root, ok := c.takeDirtyProject(); ok { |
| 221 | if project, exists, err := c.projectByRoot(c.ctx, root); err == nil && exists { |
| 222 | _ = c.ReconcileProject(c.ctx, project) |
| 223 | } else if err == nil { |
| 224 | _, _ = c.RegisterProject(c.ctx, root, filepath.Base(root)) |
| 225 | } |
| 226 | continue |
| 227 | } |
| 228 | select { |
| 229 | case <-c.ctx.Done(): |
| 230 | return |
| 231 | case <-ticker.C: |
| 232 | c.markRegisteredProjectsDirty() |
| 233 | case <-c.dirtyWake: |
| 234 | case req := <-c.queue: |
| 235 | if req.flush != nil { |
| 236 | close(req.flush) |
| 237 | continue |
| 238 | } |
| 239 | if req.events { |
| 240 | _ = c.indexEvents(c.ctx, req.projectRoot, req.taskID) |
| 241 | } else { |
| 242 | _ = c.indexSnapshot(c.ctx, req.projectRoot, req.taskID, 0) |
| 243 | } |
| 244 | } |
| 245 | } |
| 246 | } |
| 247 | |
| 248 | func (c *Catalog) takeDirtyProject() (string, bool) { |
| 249 | var root string |
| 250 | c.dirtyProjects.Range(func(key, _ any) bool { |
| 251 | root, _ = key.(string) |
| 252 | c.dirtyProjects.Delete(key) |
| 253 | return false |
| 254 | }) |
| 255 | return root, root != "" |
| 256 | } |
| 257 | |
| 258 | func (c *Catalog) markRegisteredProjectsDirty() { |
| 259 | rows, err := c.db.QueryContext(c.ctx, `SELECT project_root FROM task_projects`) |
| 260 | if err != nil { |
| 261 | return |
| 262 | } |
| 263 | defer rows.Close() |
| 264 | for rows.Next() { |
| 265 | var root string |
| 266 | if rows.Scan(&root) == nil { |
| 267 | c.dirtyProjects.Store(root, true) |
| 268 | } |
| 269 | } |
| 270 | } |
| 271 | |
| 272 | func (c *Catalog) wakeDirty() { |
| 273 | select { |
| 274 | case c.dirtyWake <- struct{}{}: |
| 275 | default: |
| 276 | } |
| 277 | } |
| 278 | |
| 279 | func (c *Catalog) Flush(ctx context.Context) error { |
| 280 | done := make(chan struct{}) |
| 281 | select { |
| 282 | case c.queue <- request{flush: done}: |
| 283 | case <-ctx.Done(): |
| 284 | return ctx.Err() |
| 285 | } |
| 286 | select { |
| 287 | case <-done: |
| 288 | return nil |
| 289 | case <-ctx.Done(): |
| 290 | return ctx.Err() |
| 291 | } |
| 292 | } |
| 293 | |
| 294 | func (c *Catalog) RegisterProject(ctx context.Context, root, label string) (Project, error) { |
| 295 | project := normalizeProject(Project{Root: root, Label: label}) |
| 296 | _, err := c.db.ExecContext(ctx, `INSERT INTO task_projects(project_key,project_root,project_label,state) VALUES(?,?,?,'pending') |
| 297 | ON CONFLICT(project_key) DO UPDATE SET project_root=excluded.project_root,project_label=excluded.project_label`, project.Key, project.Root, project.Label) |
| 298 | c.reconcileMu.Lock() |
| 299 | firstRegistration := !c.registered[project.Key] |
| 300 | c.registered[project.Key] = true |
| 301 | c.reconcileMu.Unlock() |
| 302 | if err == nil && firstRegistration { |
| 303 | c.scheduleReconcile(project) |
| 304 | } |
| 305 | return project, err |
| 306 | } |
| 307 | |
| 308 | // RequestReconcileProject forces a background authority scan even when the |
| 309 | // project was already registered by an earlier page request. |
| 310 | func (c *Catalog) RequestReconcileProject(ctx context.Context, root, label string) error { |
| 311 | project, err := c.RegisterProject(ctx, root, label) |
| 312 | if err != nil { |
| 313 | return err |
| 314 | } |
| 315 | c.scheduleReconcile(project) |
| 316 | return nil |
| 317 | } |
| 318 | |
| 319 | func (c *Catalog) scheduleReconcile(project Project) { |
| 320 | c.reconcileMu.Lock() |
| 321 | if c.reconcileDone || c.reconciling[project.Key] { |
| 322 | c.reconcileMu.Unlock() |
| 323 | return |
| 324 | } |
| 325 | c.reconciling[project.Key] = true |
| 326 | c.wg.Add(1) |
| 327 | c.reconcileMu.Unlock() |
| 328 | go func() { |
| 329 | defer func() { |
| 330 | c.reconcileMu.Lock() |
| 331 | delete(c.reconciling, project.Key) |
| 332 | c.reconcileMu.Unlock() |
| 333 | c.wg.Done() |
| 334 | }() |
| 335 | _ = c.ReconcileProject(c.ctx, project) |
| 336 | }() |
| 337 | } |
| 338 | |
| 339 | func (c *Catalog) ReconcileProject(ctx context.Context, project Project) error { |
| 340 | project = normalizeProject(project) |
| 341 | unlock := c.lockProject(project.Key) |
| 342 | defer unlock() |
| 343 | |
| 344 | tasks, err := c.store.ListTasks(ctx, project.Root) |
| 345 | if err != nil { |
| 346 | return err |
| 347 | } |
| 348 | var generation int64 |
| 349 | err = c.db.QueryRowContext(ctx, `UPDATE task_projects SET scan_generation=scan_generation+1,state='scanning',total=? WHERE project_key=? RETURNING scan_generation`, len(tasks), project.Key).Scan(&generation) |
| 350 | if err != nil { |
| 351 | return err |
| 352 | } |
| 353 | for _, task := range tasks { |
| 354 | if err := c.upsertSnapshot(ctx, project, task, generation); err != nil { |
| 355 | continue |
| 356 | } |
| 357 | } |
| 358 | now := time.Now().UnixMilli() |
| 359 | _, _ = c.db.ExecContext(ctx, `UPDATE task_snapshots SET missing_since=CASE WHEN missing_since=0 THEN ? ELSE missing_since END,health='missing' |
| 360 | WHERE project_key=? AND seen_generation<>?`, now, project.Key, generation) |
| 361 | cutoff := now - missingGrace.Milliseconds() |
| 362 | _, _ = c.db.ExecContext(ctx, `DELETE FROM task_events WHERE project_key=? AND task_id IN ( |
| 363 | SELECT task_id FROM task_snapshots WHERE project_key=? AND seen_generation<>? AND missing_since>0 AND missing_since<=? |
| 364 | )`, project.Key, project.Key, generation, cutoff) |
| 365 | _, _ = c.db.ExecContext(ctx, `DELETE FROM task_event_sources WHERE project_key=? AND task_id IN ( |
| 366 | SELECT task_id FROM task_snapshots WHERE project_key=? AND seen_generation<>? AND missing_since>0 AND missing_since<=? |
| 367 | )`, project.Key, project.Key, generation, cutoff) |
| 368 | _, _ = c.db.ExecContext(ctx, `DELETE FROM task_snapshots WHERE project_key=? AND seen_generation<>? AND missing_since>0 AND missing_since<=?`, |
| 369 | project.Key, generation, cutoff) |
| 370 | tx, beginErr := c.db.BeginTx(ctx, nil) |
| 371 | if beginErr != nil { |
| 372 | return beginErr |
| 373 | } |
| 374 | if _, err = tx.ExecContext(ctx, `UPDATE task_projects SET state='ready',indexed=?,completed_at=? WHERE project_key=?`, len(tasks), now, project.Key); err != nil { |
| 375 | _ = tx.Rollback() |
| 376 | return err |
| 377 | } |
| 378 | revision, err := bump(ctx, tx) |
| 379 | if err != nil { |
| 380 | _ = tx.Rollback() |
| 381 | return err |
| 382 | } |
| 383 | if err = tx.Commit(); err != nil { |
| 384 | return err |
| 385 | } |
| 386 | c.revision.Store(revision) |
| 387 | c.refresh(context.Background()) |
| 388 | return nil |
| 389 | } |
| 390 | |
| 391 | func (c *Catalog) indexSnapshot(ctx context.Context, root, taskID string, generation int64) error { |
| 392 | project, ok, err := c.projectByRoot(ctx, root) |
| 393 | if err != nil { |
| 394 | return err |
| 395 | } |
| 396 | if !ok { |
| 397 | project, err = c.RegisterProject(ctx, root, filepath.Base(root)) |
| 398 | if err != nil { |
| 399 | return err |
| 400 | } |
| 401 | } |
| 402 | unlock := c.lockProject(project.Key) |
| 403 | defer unlock() |
| 404 | |
| 405 | task, err := c.store.GetTask(ctx, project.Root, taskID) |
| 406 | if err != nil || task == nil { |
| 407 | return err |
| 408 | } |
| 409 | return c.upsertSnapshot(ctx, project, *task, generation) |
| 410 | } |
| 411 | |
| 412 | func (c *Catalog) upsertSnapshot(ctx context.Context, project Project, task taskmonitor.TaskSnapshot, generation int64) error { |
| 413 | b, err := json.Marshal(task) |
| 414 | if err != nil { |
| 415 | return err |
| 416 | } |
| 417 | hash := sha256.Sum256(b) |
| 418 | lease := int64(0) |
| 419 | if !task.RuntimeLeaseUntil.IsZero() { |
| 420 | lease = task.RuntimeLeaseUntil.UnixMilli() |
| 421 | } |
| 422 | tx, err := c.db.BeginTx(ctx, nil) |
| 423 | if err != nil { |
| 424 | return err |
| 425 | } |
| 426 | _, err = tx.ExecContext(ctx, `INSERT INTO task_snapshots(project_key,task_id,session_id,job_id,state,runtime_state,runtime_lease_until,version, |
| 427 | created_at,updated_at,error_code,snapshot_fingerprint,snapshot_json,health,missing_since,seen_generation) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,'ok',0,?) |
| 428 | ON CONFLICT(project_key,task_id) DO UPDATE SET session_id=excluded.session_id,job_id=excluded.job_id,state=excluded.state, |
| 429 | runtime_state=excluded.runtime_state,runtime_lease_until=excluded.runtime_lease_until,version=excluded.version,created_at=excluded.created_at, |
| 430 | updated_at=excluded.updated_at,error_code=excluded.error_code,snapshot_fingerprint=excluded.snapshot_fingerprint,snapshot_json=excluded.snapshot_json, |
| 431 | health='ok',missing_since=0,seen_generation=excluded.seen_generation`, project.Key, task.TaskID, task.SessionID, task.JobID, task.State, |
| 432 | task.RuntimeState, lease, task.Version, task.CreatedAt.UnixMilli(), task.UpdatedAt.UnixMilli(), task.ErrorCode, hex.EncodeToString(hash[:]), b, generation) |
| 433 | if err != nil { |
| 434 | _ = tx.Rollback() |
| 435 | return err |
| 436 | } |
| 437 | revision, err := bump(ctx, tx) |
| 438 | if err != nil { |
| 439 | _ = tx.Rollback() |
| 440 | return err |
| 441 | } |
| 442 | if err := tx.Commit(); err != nil { |
| 443 | return err |
| 444 | } |
| 445 | c.revision.Store(revision) |
| 446 | c.refresh(context.Background()) |
| 447 | return nil |
| 448 | } |
| 449 | |
| 450 | func (c *Catalog) indexEvents(ctx context.Context, root, taskID string) error { |
| 451 | if taskID == "" || filepath.Base(taskID) != taskID || strings.ContainsAny(taskID, `/\\`) { |
| 452 | return errors.New("invalid task id") |
| 453 | } |
| 454 | project, ok, err := c.projectByRoot(ctx, root) |
| 455 | if err != nil || !ok { |
| 456 | return err |
| 457 | } |
| 458 | unlock := c.lockProject(project.Key) |
| 459 | defer unlock() |
| 460 | |
| 461 | path := filepath.Join(project.Root, ".reasonix", "tasks", taskID, "events.jsonl") |
| 462 | info, statErr := os.Stat(path) |
| 463 | if errors.Is(statErr, os.ErrNotExist) { |
| 464 | return nil |
| 465 | } |
| 466 | if statErr != nil { |
| 467 | return statErr |
| 468 | } |
| 469 | fingerprint := fmt.Sprintf("%d:%d", info.Size(), info.ModTime().UnixNano()) |
| 470 | var oldPath, oldFingerprint string |
| 471 | var oldSize, offset int64 |
| 472 | err = c.db.QueryRowContext(ctx, `SELECT path,size,indexed_offset,fingerprint FROM task_event_sources WHERE project_key=? AND task_id=?`, |
| 473 | project.Key, taskID).Scan(&oldPath, &oldSize, &offset, &oldFingerprint) |
| 474 | reset := errors.Is(err, sql.ErrNoRows) || oldPath != path || info.Size() < oldSize || info.Size() == oldSize && fingerprint != oldFingerprint |
| 475 | if err != nil && !errors.Is(err, sql.ErrNoRows) { |
| 476 | return err |
| 477 | } |
| 478 | if !reset && info.Size() == oldSize && fingerprint == oldFingerprint { |
| 479 | return nil |
| 480 | } |
| 481 | if reset { |
| 482 | offset = 0 |
| 483 | } |
| 484 | tail, err := c.store.ReadEventTail(ctx, project.Root, taskID, offset) |
| 485 | if err != nil { |
| 486 | return err |
| 487 | } |
| 488 | reset = reset || tail.Reset |
| 489 | tx, err := c.db.BeginTx(ctx, nil) |
| 490 | if err != nil { |
| 491 | return err |
| 492 | } |
| 493 | if reset { |
| 494 | if _, err := tx.ExecContext(ctx, `DELETE FROM task_events WHERE project_key=? AND task_id=?`, project.Key, taskID); err != nil { |
| 495 | _ = tx.Rollback() |
| 496 | return err |
| 497 | } |
| 498 | } |
| 499 | for _, event := range tail.Items { |
| 500 | b, _ := json.Marshal(event) |
| 501 | if _, err := tx.ExecContext(ctx, `INSERT OR REPLACE INTO task_events(project_key,task_id,sequence,timestamp,event_type,state,runtime_state,event_json) |
| 502 | VALUES(?,?,?,?,?,?,?,?)`, project.Key, taskID, event.Sequence, event.Timestamp.UnixMilli(), event.EventType, event.State, event.RuntimeState, b); err != nil { |
| 503 | _ = tx.Rollback() |
| 504 | return err |
| 505 | } |
| 506 | } |
| 507 | _, err = tx.ExecContext(ctx, `INSERT INTO task_event_sources(project_key,task_id,path,size,indexed_offset,fingerprint,state,error) |
| 508 | VALUES(?,?,?,?,?,?,'ready','') ON CONFLICT(project_key,task_id) DO UPDATE SET path=excluded.path,size=excluded.size, |
| 509 | indexed_offset=excluded.indexed_offset,fingerprint=excluded.fingerprint,state='ready',error=''`, project.Key, taskID, path, |
| 510 | info.Size(), tail.NextOffset, fingerprint) |
| 511 | if err != nil { |
| 512 | _ = tx.Rollback() |
| 513 | return err |
| 514 | } |
| 515 | return tx.Commit() |
| 516 | } |
| 517 | |
| 518 | func (c *Catalog) ListPage(ctx context.Context, req PageRequest) (Page, error) { |
| 519 | status := c.Status() |
| 520 | out := Page{Items: []Item{}, Revision: c.revision.Load(), Partial: status.Pending > 0, Status: status} |
| 521 | limit := req.Limit |
| 522 | if limit <= 0 { |
| 523 | limit = DefaultLimit |
| 524 | } |
| 525 | if limit > MaxLimit { |
| 526 | limit = MaxLimit |
| 527 | } |
| 528 | cur, err := decodeCursor(req.Cursor) |
| 529 | if err != nil { |
| 530 | return out, err |
| 531 | } |
| 532 | if cur != nil && cur.Revision != out.Revision { |
| 533 | out.StaleCursor = true |
| 534 | return out, nil |
| 535 | } |
| 536 | where := []string{`s.health='ok'`, `s.missing_since=0`} |
| 537 | args := []any{} |
| 538 | if len(req.ProjectKeys) > 0 { |
| 539 | parts := make([]string, len(req.ProjectKeys)) |
| 540 | for i, key := range req.ProjectKeys { |
| 541 | parts[i] = "?" |
| 542 | args = append(args, key) |
| 543 | } |
| 544 | where = append(where, `s.project_key IN (`+strings.Join(parts, ",")+`)`) |
| 545 | } |
| 546 | if req.SessionID != "" { |
| 547 | where = append(where, `s.session_id=?`) |
| 548 | args = append(args, req.SessionID) |
| 549 | } |
| 550 | if len(req.States) > 0 { |
| 551 | parts := make([]string, len(req.States)) |
| 552 | for i, state := range req.States { |
| 553 | parts[i] = "?" |
| 554 | args = append(args, state) |
| 555 | } |
| 556 | where = append(where, `s.state IN (`+strings.Join(parts, ",")+`)`) |
| 557 | } |
| 558 | if query := strings.ToLower(strings.TrimSpace(req.Query)); query != "" { |
| 559 | where = append(where, `(lower(s.task_id) LIKE ? OR lower(s.session_id) LIKE ? OR lower(s.error_code) LIKE ?)`) |
| 560 | like := "%" + query + "%" |
| 561 | args = append(args, like, like, like) |
| 562 | } |
| 563 | if cur != nil { |
| 564 | where = append(where, `(s.updated_at<? OR (s.updated_at=? AND s.project_key>?) OR (s.updated_at=? AND s.project_key=? AND s.task_id>?))`) |
| 565 | args = append(args, cur.Updated, cur.Updated, cur.Project, cur.Updated, cur.Project, cur.Task) |
| 566 | } |
| 567 | args = append(args, limit+1) |
| 568 | rows, err := c.db.QueryContext(ctx, `SELECT s.project_key,p.project_label,s.snapshot_json FROM task_snapshots s JOIN task_projects p ON p.project_key=s.project_key |
| 569 | WHERE `+strings.Join(where, ` AND `)+` ORDER BY s.updated_at DESC,s.project_key,s.task_id LIMIT ?`, args...) |
| 570 | if err != nil { |
| 571 | return out, err |
| 572 | } |
| 573 | defer rows.Close() |
| 574 | for rows.Next() { |
| 575 | var item Item |
| 576 | var raw []byte |
| 577 | if err := rows.Scan(&item.ProjectKey, &item.ProjectLabel, &raw); err != nil { |
| 578 | return out, err |
| 579 | } |
| 580 | if json.Unmarshal(raw, &item.Task) != nil { |
| 581 | continue |
| 582 | } |
| 583 | item.Task.ReconcileRuntime(time.Now()) |
| 584 | out.Items = append(out.Items, item) |
| 585 | } |
| 586 | if len(out.Items) > limit { |
| 587 | out.Items = out.Items[:limit] |
| 588 | last := out.Items[len(out.Items)-1] |
| 589 | out.NextCursor = encodeCursor(cursor{Revision: out.Revision, Updated: last.Task.UpdatedAt.UnixMilli(), Project: last.ProjectKey, Task: last.Task.TaskID}) |
| 590 | } |
| 591 | return out, rows.Err() |
| 592 | } |
| 593 | |
| 594 | func (c *Catalog) ListEventPage(ctx context.Context, projectKey, taskID string, after, limit int) (EventPage, error) { |
| 595 | out := EventPage{Items: []taskmonitor.TaskEvent{}, NextSequence: after} |
| 596 | project, ok, err := c.Project(ctx, projectKey) |
| 597 | if err != nil || !ok { |
| 598 | return out, err |
| 599 | } |
| 600 | if err := c.indexEvents(ctx, project.Root, taskID); err != nil { |
| 601 | out.Partial = true |
| 602 | return out, nil |
| 603 | } |
| 604 | if limit <= 0 { |
| 605 | limit = DefaultLimit |
| 606 | } |
| 607 | if limit > MaxLimit { |
| 608 | limit = MaxLimit |
| 609 | } |
| 610 | rows, err := c.db.QueryContext(ctx, `SELECT event_json FROM task_events WHERE project_key=? AND task_id=? AND sequence>? ORDER BY sequence LIMIT ?`, projectKey, taskID, after, limit) |
| 611 | if err != nil { |
| 612 | return out, err |
| 613 | } |
| 614 | defer rows.Close() |
| 615 | for rows.Next() { |
| 616 | var raw []byte |
| 617 | var event taskmonitor.TaskEvent |
| 618 | if err := rows.Scan(&raw); err != nil { |
| 619 | return out, err |
| 620 | } |
| 621 | if json.Unmarshal(raw, &event) == nil { |
| 622 | out.Items = append(out.Items, event) |
| 623 | out.NextSequence = event.Sequence |
| 624 | } |
| 625 | } |
| 626 | return out, rows.Err() |
| 627 | } |
| 628 | |
| 629 | func (c *Catalog) Project(ctx context.Context, key string) (Project, bool, error) { |
| 630 | var project Project |
| 631 | err := c.db.QueryRowContext(ctx, `SELECT project_key,project_root,project_label FROM task_projects WHERE project_key=?`, key).Scan(&project.Key, &project.Root, &project.Label) |
| 632 | if errors.Is(err, sql.ErrNoRows) { |
| 633 | return project, false, nil |
| 634 | } |
| 635 | return project, err == nil, err |
| 636 | } |
| 637 | |
| 638 | func (c *Catalog) projectByRoot(ctx context.Context, root string) (Project, bool, error) { |
| 639 | return c.Project(ctx, ProjectKey(root)) |
| 640 | } |
| 641 | |
| 642 | func bump(ctx context.Context, tx *sql.Tx) (uint64, error) { |
| 643 | if _, err := tx.ExecContext(ctx, `UPDATE task_state SET revision=revision+1 WHERE id=1`); err != nil { |
| 644 | return 0, err |
| 645 | } |
| 646 | var revision uint64 |
| 647 | err := tx.QueryRowContext(ctx, `SELECT revision FROM task_state WHERE id=1`).Scan(&revision) |
| 648 | return revision, err |
| 649 | } |
| 650 | |
| 651 | func encodeCursor(value cursor) string { |
| 652 | b, _ := json.Marshal(value) |
| 653 | return base64.RawURLEncoding.EncodeToString(b) |
| 654 | } |
| 655 | |
| 656 | func decodeCursor(value string) (*cursor, error) { |
| 657 | if strings.TrimSpace(value) == "" { |
| 658 | return nil, nil |
| 659 | } |
| 660 | b, err := base64.RawURLEncoding.DecodeString(value) |
| 661 | if err != nil { |
| 662 | return nil, fmt.Errorf("invalid task cursor: %w", err) |
| 663 | } |
| 664 | var out cursor |
| 665 | if json.Unmarshal(b, &out) != nil || out.Task == "" { |
| 666 | return nil, errors.New("invalid task cursor") |
| 667 | } |
| 668 | return &out, nil |
| 669 | } |
| 670 | |
| 671 | func (c *Catalog) refresh(ctx context.Context) { |
| 672 | var indexed, total, pending, failed int64 |
| 673 | _ = c.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM task_snapshots WHERE health='ok'`).Scan(&indexed) |
| 674 | _ = c.db.QueryRowContext(ctx, `SELECT COALESCE(SUM(total),0) FROM task_projects`).Scan(&total) |
| 675 | _ = c.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM task_projects WHERE state<>'ready'`).Scan(&pending) |
| 676 | _ = c.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM task_snapshots WHERE health='corrupt'`).Scan(&failed) |
| 677 | c.statusMu.Lock() |
| 678 | c.status.Revision, c.status.Indexed, c.status.Total, c.status.Pending, c.status.Failed = c.revision.Load(), indexed, total, pending, failed |
| 679 | c.statusMu.Unlock() |
| 680 | } |
| 681 | |
| 682 | func (c *Catalog) Status() Status { |
| 683 | c.statusMu.RLock() |
| 684 | defer c.statusMu.RUnlock() |
| 685 | return c.status |
| 686 | } |
| 687 | |
| 688 | func (c *Catalog) Close(ctx context.Context) error { |
| 689 | if c == nil { |
| 690 | return nil |
| 691 | } |
| 692 | c.closeOnce.Do(func() { |
| 693 | c.closing.Store(true) |
| 694 | c.cancel() |
| 695 | c.reconcileMu.Lock() |
| 696 | c.reconcileDone = true |
| 697 | c.reconcileMu.Unlock() |
| 698 | go func() { |
| 699 | c.wg.Wait() |
| 700 | c.closeErr = c.db.Close() |
| 701 | close(c.closeDone) |
| 702 | }() |
| 703 | }) |
| 704 | select { |
| 705 | case <-c.closeDone: |
| 706 | return c.closeErr |
| 707 | case <-ctx.Done(): |
| 708 | return ctx.Err() |
| 709 | } |
| 710 | } |
| 711 |