| 1 | package memoria |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "log/slog" |
| 7 | "sync" |
| 8 | "time" |
| 9 | |
| 10 | "gorm.io/gorm" |
| 11 | |
| 12 | "amadan.net/oficina/memoria/internal/memoria/chunk" |
| 13 | "amadan.net/oficina/memoria/internal/memoria/notion" |
| 14 | "amadan.net/oficina/memoria/internal/memoria/seal" |
| 15 | ) |
| 16 | |
| 17 | // SourceClient is what a sync needs from a source. An interface so the |
| 18 | // loop can be driven against a fake without the app package importing an |
| 19 | // HTTP client — which the network gate forbids it anyway. |
| 20 | type SourceClient interface { |
| 21 | // Roots returns the true roots and, separately, the orphans — |
| 22 | // results whose parent search did not return. See notion.Roots. |
| 23 | Roots(ctx context.Context) (roots, orphans []notion.Ref, err error) |
| 24 | Fetch(ctx context.Context, ref notion.Ref) (*notion.Content, error) |
| 25 | } |
| 26 | |
| 27 | // Syncer runs imports. |
| 28 | type Syncer struct { |
| 29 | store *Store |
| 30 | logger *slog.Logger |
| 31 | // dial builds a client from a token. Swapped in tests. |
| 32 | dial func(token string) (SourceClient, error) |
| 33 | } |
| 34 | |
| 35 | // NewSyncer wires a syncer against the real Notion API. |
| 36 | func NewSyncer(store *Store, logger *slog.Logger) *Syncer { |
| 37 | return &Syncer{store: store, logger: logger, dial: func(token string) (SourceClient, error) { |
| 38 | return notion.New(token) |
| 39 | }} |
| 40 | } |
| 41 | |
| 42 | // Run imports everything reachable from the connector's roots. |
| 43 | // |
| 44 | // ONE PASS, not the two phases an earlier design specified. With |
| 45 | // traversal, enumeration and extraction are the same API call — fetching |
| 46 | // a page returns its body AND the child pages beneath it — so separating |
| 47 | // them would double every request against a limit of three per second. |
| 48 | // The ledger is still written per page, so the ingest work already done |
| 49 | // survives an interruption; what a resumed run repeats is the traversal, |
| 50 | // because a page's children cannot be discovered without asking for them |
| 51 | // and a child's own edits do not change its parent's edit time. |
| 52 | // |
| 53 | // Errors divide in two, and the division is load-bearing: |
| 54 | // |
| 55 | // - A traversal failure below a page (rate limit exhausted, a broken |
| 56 | // cursor, an unreadable body) is FATAL. It means the pages beneath |
| 57 | // that point were never enumerated, and a run that continued would |
| 58 | // report "done" while missing a whole subtree — an absence produced |
| 59 | // by a failure, which is the shape this codebase ranks worst. |
| 60 | // - A single page that has gone (404 between enumeration and fetch) is |
| 61 | // recorded against that page and the run carries on. One placeholder |
| 62 | // must not block an import, but it must not vanish either. |
| 63 | func (s *Syncer) Run(ctx context.Context, connectorID, runID int64) { |
| 64 | token, err := s.store.ConnectorToken(ctx, "notion") |
| 65 | if err != nil { |
| 66 | s.fail(ctx, runID, "unauthorized") |
| 67 | return |
| 68 | } |
| 69 | client, err := s.dial(token) |
| 70 | if err != nil { |
| 71 | s.fail(ctx, runID, "unauthorized") |
| 72 | return |
| 73 | } |
| 74 | s.RunWith(ctx, client, connectorID, runID) |
| 75 | } |
| 76 | |
| 77 | // runStats is the live count of what a run has done. |
| 78 | // |
| 79 | // Guarded, because the heartbeat goroutine reads it while the walk |
| 80 | // writes it — and it is read for a reason: an earlier version wrote the |
| 81 | // counters only at the very end, so a thirty-minute import rendered |
| 82 | // "0 so far" for its whole duration. A count maintained nowhere and |
| 83 | // displayed as zero is this codebase's second broken rule, on the status |
| 84 | // page written to honour it. |
| 85 | type runStats struct { |
| 86 | mu sync.Mutex |
| 87 | fetched, skipped, empty, failed, unread, seen int |
| 88 | } |
| 89 | |
| 90 | func (r *runStats) add(f func(*runStats)) { |
| 91 | r.mu.Lock() |
| 92 | defer r.mu.Unlock() |
| 93 | f(r) |
| 94 | } |
| 95 | |
| 96 | func (r *runStats) snapshot() map[string]any { |
| 97 | r.mu.Lock() |
| 98 | defer r.mu.Unlock() |
| 99 | return map[string]any{ |
| 100 | "fetched": r.fetched, "skipped": r.skipped, "empty": r.empty, |
| 101 | "failed_pages": r.failed, "unread_blocks": r.unread, |
| 102 | } |
| 103 | } |
| 104 | |
| 105 | // beatEvery is how often a run says it is still alive AND publishes its |
| 106 | // counters. A DURATION, not a page count: pages take wildly different |
| 107 | // times — one request per nested block, each paced at 350ms, plus an |
| 108 | // honoured Retry-After — so "every 25 pages" could exceed the staleness |
| 109 | // window while the run was perfectly healthy. StartRun would then reclaim |
| 110 | // a LIVE run and start a second goroutine on the same ledger. Found by a |
| 111 | // code review. |
| 112 | const beatEvery = 20 * time.Second |
| 113 | |
| 114 | // startBeat publishes the heartbeat and the counters until stopped. |
| 115 | func (s *Syncer) startBeat(ctx context.Context, runID int64, st *runStats) func() { |
| 116 | done := make(chan struct{}) |
| 117 | go func() { |
| 118 | t := time.NewTicker(beatEvery) |
| 119 | defer t.Stop() |
| 120 | for { |
| 121 | select { |
| 122 | case <-done: |
| 123 | return |
| 124 | case <-t.C: |
| 125 | patch := st.snapshot() |
| 126 | patch["heartbeat"] = time.Now().UTC() |
| 127 | if err := s.store.UpdateRun(ctx, runID, patch); err != nil { |
| 128 | s.logger.Error("sync heartbeat", "run", runID, "err", err) |
| 129 | } |
| 130 | } |
| 131 | } |
| 132 | }() |
| 133 | return func() { close(done) } |
| 134 | } |
| 135 | |
| 136 | // RunWith is Run against a supplied client. |
| 137 | func (s *Syncer) RunWith(ctx context.Context, client SourceClient, connectorID, runID int64) { |
| 138 | roots, orphans, err := client.Roots(ctx) |
| 139 | if err != nil { |
| 140 | s.fail(ctx, runID, string(notion.CodeOf(err))) |
| 141 | return |
| 142 | } |
| 143 | if err := s.store.UpdateRun(ctx, runID, map[string]any{ |
| 144 | "roots": len(roots), "state": RunFetching, |
| 145 | }); err != nil { |
| 146 | s.fail(ctx, runID, "database") |
| 147 | return |
| 148 | } |
| 149 | |
| 150 | st := &runStats{} |
| 151 | stop := s.startBeat(ctx, runID, st) |
| 152 | defer stop() |
| 153 | |
| 154 | visited := map[string]bool{} |
| 155 | |
| 156 | // The real tree first, so every page that HAS an ancestor is reached |
| 157 | // through it and inherits the right path. |
| 158 | if !s.walk(ctx, client, connectorID, runID, roots, visited, st) { |
| 159 | return |
| 160 | } |
| 161 | // Then the orphans — results whose parent search did not return, most |
| 162 | // often a page nested inside a toggle, whose parent is a block. They |
| 163 | // go last and skip anything already visited, so an orphan can never |
| 164 | // claim a page the traversal would have given real ancestry to. |
| 165 | if !s.walk(ctx, client, connectorID, runID, orphans, visited, st) { |
| 166 | return |
| 167 | } |
| 168 | |
| 169 | now := time.Now().UTC() |
| 170 | patch := st.snapshot() |
| 171 | st.mu.Lock() |
| 172 | patch["total"] = st.seen |
| 173 | st.mu.Unlock() |
| 174 | patch["state"] = RunDone |
| 175 | patch["finished_at"] = now |
| 176 | patch["heartbeat"] = now |
| 177 | if err := s.store.UpdateRun(ctx, runID, patch); err != nil { |
| 178 | s.logger.Error("sync finish", "run", runID, "err", err) |
| 179 | } |
| 180 | // Counts only. No title, no path, no page id — AGENTS.md's rule, and |
| 181 | // the reason this line carries nothing that came from the workspace. |
| 182 | s.logger.Info("sync finished", "run", runID, "pages", patch["total"], |
| 183 | "fetched", patch["fetched"], "skipped", patch["skipped"], |
| 184 | "empty", patch["empty"], "failed", patch["failed_pages"], |
| 185 | "unread_blocks", patch["unread_blocks"]) |
| 186 | } |
| 187 | |
| 188 | // walk breadth-first from seeds. It reports false when the run has |
| 189 | // failed, having already marked it. |
| 190 | func (s *Syncer) walk(ctx context.Context, client SourceClient, connectorID, runID int64, |
| 191 | seeds []notion.Ref, visited map[string]bool, st *runStats) bool { |
| 192 | |
| 193 | queue := append([]notion.Ref{}, seeds...) |
| 194 | for len(queue) > 0 { |
| 195 | ref := queue[0] |
| 196 | queue = queue[1:] |
| 197 | if visited[ref.ID] { |
| 198 | continue |
| 199 | } |
| 200 | visited[ref.ID] = true |
| 201 | st.add(func(r *runStats) { r.seen++ }) |
| 202 | |
| 203 | // A data source is a CONTAINER, not a document. It gets no ledger |
| 204 | // row: one there would sit "pending" forever, be counted in the |
| 205 | // total, and be counted as "already up to date" on every later |
| 206 | // run — so the page said "1 of 2 imported" permanently. |
| 207 | container := ref.Kind == notion.KindDataSource |
| 208 | if !container { |
| 209 | // Record the coordinate BEFORE fetching, so a page that then |
| 210 | // fails to read has somewhere to be named rather than being |
| 211 | // absent. |
| 212 | if err := s.store.SeeSourcePage(ctx, connectorID, ref.ID, ref.Title, ref.Path, ref.EditedTime); err != nil { |
| 213 | s.logger.Error("sync ledger write", "run", runID, "err", err) |
| 214 | s.fail(ctx, runID, "database") |
| 215 | return false |
| 216 | } |
| 217 | } |
| 218 | |
| 219 | content, err := client.Fetch(ctx, ref) |
| 220 | if err != nil { |
| 221 | if gone(err) { |
| 222 | st.add(func(r *runStats) { r.failed++ }) |
| 223 | if !container { |
| 224 | s.markGone(ctx, connectorID, ref, string(notion.CodeOf(err))) |
| 225 | } |
| 226 | continue |
| 227 | } |
| 228 | // Not survivable: the subtree below here was never seen. |
| 229 | s.fail(ctx, runID, string(notion.CodeOf(err))) |
| 230 | return false |
| 231 | } |
| 232 | queue = append(queue, content.Children...) |
| 233 | st.add(func(r *runStats) { r.unread += content.UnreadBlocks }) |
| 234 | |
| 235 | if container { |
| 236 | continue |
| 237 | } |
| 238 | |
| 239 | state, docID, err := s.ingest(ctx, connectorID, ref, content) |
| 240 | if err != nil { |
| 241 | st.add(func(r *runStats) { r.failed++ }) |
| 242 | s.markGone(ctx, connectorID, ref, "extract_failed") |
| 243 | continue |
| 244 | } |
| 245 | switch state { |
| 246 | case PageEmpty: |
| 247 | st.add(func(r *runStats) { r.empty++ }) |
| 248 | case "": |
| 249 | st.add(func(r *runStats) { r.skipped++ }) |
| 250 | continue |
| 251 | default: |
| 252 | st.add(func(r *runStats) { r.fetched++ }) |
| 253 | } |
| 254 | s.markByPage(ctx, connectorID, ref, state, "", docID, ref.EditedTime) |
| 255 | } |
| 256 | return true |
| 257 | } |
| 258 | |
| 259 | // ingest stores one page, or reports that it did not need storing. |
| 260 | // |
| 261 | // Returns "" when the page was already up to date. A data source is a |
| 262 | // container rather than a document and is never stored. |
| 263 | func (s *Syncer) ingest(ctx context.Context, connectorID int64, ref notion.Ref, content *notion.Content) (string, int64, error) { |
| 264 | if ref.Kind == notion.KindDataSource { |
| 265 | return "", 0, nil |
| 266 | } |
| 267 | upToDate, err := s.upToDate(ctx, connectorID, ref) |
| 268 | if err != nil { |
| 269 | return "", 0, err |
| 270 | } |
| 271 | if upToDate { |
| 272 | return "", 0, nil |
| 273 | } |
| 274 | id, err := s.store.Ingest(ctx, IngestSpec{ |
| 275 | Source: "notion", |
| 276 | Key: ref.ID, |
| 277 | Title: ref.Title, |
| 278 | Body: content.Markdown, |
| 279 | Path: ref.Path, |
| 280 | // On, because an empty page is a fact about the source. See |
| 281 | // IngestSpec.AllowTitleOnly. |
| 282 | AllowTitleOnly: true, |
| 283 | }) |
| 284 | if err != nil { |
| 285 | // ErrIndexLag means the row COMMITTED and only the in-memory |
| 286 | // index is behind. The document exists, is in the list, and comes |
| 287 | // back at the next rebuild, which is every boot. Recording that |
| 288 | // as a failure — and wiping the id — would be telling someone a |
| 289 | // document is missing when it is right there. |
| 290 | if errors.Is(err, ErrIndexLag) { |
| 291 | s.logger.Error("index refresh failed after commit", "source", "notion", "err", err) |
| 292 | if content.Markdown == "" { |
| 293 | return PageEmpty, id, nil |
| 294 | } |
| 295 | return PageFetched, id, nil |
| 296 | } |
| 297 | return "", 0, err |
| 298 | } |
| 299 | if content.Markdown == "" { |
| 300 | return PageEmpty, id, nil |
| 301 | } |
| 302 | return PageFetched, id, nil |
| 303 | } |
| 304 | |
| 305 | // upToDate reports whether the ledger already holds this page at this |
| 306 | // edit time AND this ancestry. |
| 307 | // |
| 308 | // Both, because an ancestor rename moves the ancestry without moving the |
| 309 | // page's own edit time — so a check on the edit time alone leaves every |
| 310 | // descendant of a renamed page stale forever, with nothing upstream ever |
| 311 | // signalling it. |
| 312 | func (s *Syncer) upToDate(ctx context.Context, connectorID int64, ref notion.Ref) (bool, error) { |
| 313 | rows, err := s.store.pageRow(ctx, connectorID, ref.ID) |
| 314 | if err != nil || rows == nil { |
| 315 | return false, err |
| 316 | } |
| 317 | if rows.State != PageFetched && rows.State != PageEmpty { |
| 318 | return false, nil |
| 319 | } |
| 320 | if rows.IngestedEdited != ref.EditedTime { |
| 321 | return false, nil |
| 322 | } |
| 323 | want, err := s.store.pathPrint(ref.Path, ref.Title) |
| 324 | if err != nil { |
| 325 | return false, err |
| 326 | } |
| 327 | return rows.IngestedPath == want, nil |
| 328 | } |
| 329 | |
| 330 | // markGone records a failure WITHOUT erasing what memoria already holds. |
| 331 | // |
| 332 | // An earlier version routed failures through markByPage with a zero |
| 333 | // document id and an empty edit time, so a page that had imported fine |
| 334 | // and then 404'd on a later run had its record wiped: "Imported" dropped, |
| 335 | // and the next run had to fetch and re-ingest a document that was never |
| 336 | // gone. Found by a code review. |
| 337 | func (s *Syncer) markGone(ctx context.Context, connectorID int64, ref notion.Ref, code string) { |
| 338 | row, err := s.store.pageRow(ctx, connectorID, ref.ID) |
| 339 | if err != nil { |
| 340 | s.logger.Error("sync read ledger row", "err", err) |
| 341 | return |
| 342 | } |
| 343 | if row == nil { |
| 344 | return |
| 345 | } |
| 346 | // Keep document_id, ingested_edited and ingested_path exactly as they |
| 347 | // were; only the state and the code move. |
| 348 | if err := s.store.MarkPage(ctx, row.ID, PageFailed, code, |
| 349 | row.DocumentID, row.IngestedEdited, row.IngestedPath); err != nil { |
| 350 | s.logger.Error("sync mark page", "err", err) |
| 351 | } |
| 352 | } |
| 353 | |
| 354 | func (s *Syncer) markByPage(ctx context.Context, connectorID int64, ref notion.Ref, state, code string, docID int64, edited string) { |
| 355 | row, err := s.store.pageRow(ctx, connectorID, ref.ID) |
| 356 | if err != nil { |
| 357 | s.logger.Error("sync read ledger row", "err", err) |
| 358 | return |
| 359 | } |
| 360 | if row == nil { |
| 361 | return |
| 362 | } |
| 363 | print, err := s.store.pathPrint(ref.Path, ref.Title) |
| 364 | if err != nil { |
| 365 | s.logger.Error("sync path print", "err", err) |
| 366 | return |
| 367 | } |
| 368 | if edited == "" { |
| 369 | print = "" |
| 370 | } |
| 371 | if err := s.store.MarkPage(ctx, row.ID, state, code, docID, edited, print); err != nil { |
| 372 | s.logger.Error("sync mark page", "err", err) |
| 373 | } |
| 374 | } |
| 375 | |
| 376 | func (s *Syncer) fail(ctx context.Context, runID int64, code string) { |
| 377 | now := time.Now().UTC() |
| 378 | if err := s.store.UpdateRun(ctx, runID, map[string]any{ |
| 379 | "state": RunFailed, "error_code": code, "finished_at": now, |
| 380 | }); err != nil { |
| 381 | s.logger.Error("sync fail", "run", runID, "err", err) |
| 382 | } |
| 383 | s.logger.Error("sync failed", "run", runID, "code", code) |
| 384 | } |
| 385 | |
| 386 | func asNotionErr(err error, target **notion.Error) bool { return errors.As(err, target) } |
| 387 | |
| 388 | // gone reports whether an error is about one page rather than about the |
| 389 | // traversal. Only a 404 qualifies: everything else means we do not know |
| 390 | // what lies below, and continuing would report a complete import. |
| 391 | func gone(err error) bool { |
| 392 | var e *notion.Error |
| 393 | if !asNotionErr(err, &e) { |
| 394 | return false |
| 395 | } |
| 396 | return e.Status == 404 |
| 397 | } |
| 398 | |
| 399 | // pageRow reads one ledger row by coordinate. |
| 400 | func (s *Store) pageRow(ctx context.Context, connectorID int64, pageID string) (*SourcePage, error) { |
| 401 | print, err := s.sealer.Fingerprint(seal.PurposeCoordinate, []byte("page\x00"+pageID)) |
| 402 | if err != nil { |
| 403 | return nil, err |
| 404 | } |
| 405 | var row SourcePage |
| 406 | res := s.db.G.WithContext(ctx).Where("connector_id = ? AND page_print = ?", connectorID, print).First(&row) |
| 407 | if res.Error != nil { |
| 408 | // NOT-FOUND is not an error; anything else is. This returned |
| 409 | // (nil, nil) for both, so a database failure looked exactly like |
| 410 | // "memoria has never seen this page" — the caller then skipped |
| 411 | // its mark silently and the ledger under-reported with nothing |
| 412 | // logged. The plausible-empty-result shape, in a helper. |
| 413 | if errors.Is(res.Error, gorm.ErrRecordNotFound) { |
| 414 | return nil, nil |
| 415 | } |
| 416 | return nil, res.Error |
| 417 | } |
| 418 | return &row, nil |
| 419 | } |
| 420 | |
| 421 | // pathPrint is the keyed fingerprint of an ancestry plus a title, so a |
| 422 | // move or a rename is detectable without opening the row. |
| 423 | func (s *Store) pathPrint(path []string, title string) (string, error) { |
| 424 | pathStr := chunk.Chunk{Path: path}.PathString() |
| 425 | return s.sealer.Fingerprint(seal.PurposePath, []byte("path\x00"+pathStr+"\x00"+title)) |
| 426 | } |
| 427 | |
amadan