Skip to content

Commit 24c5734

Browse files
altaywtfaltaywtf
committed
feat(sync): persist intercom tail state
1 parent c741758 commit 24c5734

9 files changed

Lines changed: 495 additions & 13 deletions

File tree

‎README.md‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -61,6 +61,14 @@ For a bounded recent crawl, prefer a short window and explicit limit:
6161
FINCRAWL_HOME=/tmp/fincrawl-live-smoke go run ./cmd/fincrawl sync --updated-since 2h --limit 5 --json
6262
```
6363

64+
When a limited or interrupted updated-since run leaves an active window in local
65+
SQLite, fresh updated-since runs are refused until the active window is
66+
continued with:
67+
68+
```bash
69+
FINCRAWL_HOME=/tmp/fincrawl-live-smoke go run ./cmd/fincrawl sync --resume --json
70+
```
71+
6472
The default Intercom API version is `2.15`. Set
6573
`FINCRAWL_INTERCOM_BASE_URL` only for a regional Intercom API host and
6674
`FINCRAWL_INTERCOM_VERSION` only when intentionally testing another supported

‎docs/plans/bootstrap-mvp.md‎

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@ fincrawl metadata --json
2525
fincrawl status --json
2626
fincrawl sync --fixture testdata/synthetic
2727
fincrawl sync --updated-since 2h
28+
fincrawl sync --resume
2829
fincrawl sync --conversation <id>
2930
fincrawl search "billing refund" --json
3031
fincrawl archive --fixture testdata/synthetic --recipient <age-recipient> --out <tmp>.jsonl.zst.age
@@ -115,8 +116,17 @@ development path. It should exercise the same store and search code paths as
115116
provider sync.
116117

117118
`sync --updated-since` uses the Intercom search shape described in
118-
[Intercom API reference](../references/intercom-api.md). It is allowed to fail
119-
with a clear missing-credential diagnostic when no local token is present.
119+
[Intercom API reference](../references/intercom-api.md). It persists an active
120+
sync window before provider reads, writes hydrated conversations as they arrive,
121+
and leaves resumable state when `--limit` stops before the window completes. It
122+
is allowed to fail with a clear missing-credential diagnostic when no local
123+
token is present.
124+
125+
`sync --resume` continues the active Intercom updated-since window from
126+
`sync_state`. It reuses the saved page cursor and last processed provider ID so
127+
an interrupted or intentionally bounded run does not skip rows in the active
128+
window. A completed window clears active state and advances the high-water mark.
129+
Fresh `sync --updated-since` runs are refused while active state exists.
120130

121131
`sync --conversation` hydrates one provider conversation by ID and writes the
122132
same normalized rows and raw blobs as incremental sync. It is the exact-refresh

‎docs/specs/intercom-archive-mvp.md‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@ fincrawl doctor
3333
fincrawl metadata --json
3434
fincrawl status --json
3535
fincrawl sync --updated-since 30d
36+
fincrawl sync --resume
3637
fincrawl sync --conversation <id>
3738
fincrawl search "billing refund" --json
3839
fincrawl conversations --fin --unresolved
@@ -183,6 +184,10 @@ sync. Query on `updated_at`, sort deterministically, page with
183184
`pages.next.starting_after`, and use a bounded lookback window. Store enough sync
184185
state to recover from interrupted runs and timestamp collisions: at minimum the
185186
successful high-water mark plus the last processed provider ID or page cursor.
187+
Bounded or interrupted active windows resume explicitly with
188+
`fincrawl sync --resume`; completing the window clears active state and advances
189+
the high-water mark. Starting a fresh tail window while active state exists is
190+
rejected unless an explicit abandon or force path is added later.
186191

187192
Treat conversation parts as hydration data. Search/list endpoints identify
188193
changed conversations; retrieving a single conversation returns its

‎internal/cli/cli.go‎

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -168,6 +168,7 @@ type syncCmd struct {
168168
Fixture string `help:"Import synthetic fixture directory."`
169169
UpdatedSince string `name:"updated-since" help:"Sync provider conversations updated since a duration or timestamp."`
170170
Conversation string `help:"Hydrate one provider conversation ID."`
171+
Resume bool `help:"Resume an interrupted Intercom updated-since sync window."`
171172
Limit int `help:"Maximum provider conversations to hydrate for --updated-since. Use 0 for no limit." default:"50"`
172173
JSON bool `help:"Print JSON output."`
173174
}
@@ -199,7 +200,7 @@ func (cmd syncCmd) Run(ctx commandContext) error {
199200
}
200201
return writeMaybeJSON(ctx.stdout, cmd.JSON, result)
201202
}
202-
if cmd.UpdatedSince != "" || cmd.Conversation != "" {
203+
if cmd.UpdatedSince != "" || cmd.Conversation != "" || cmd.Resume {
203204
if config.IntercomToken() == "" {
204205
return fmt.Errorf("missing %s for live Intercom sync", config.EnvIntercomCred)
205206
}
@@ -231,6 +232,8 @@ func (cmd syncCmd) Run(ctx commandContext) error {
231232
var result store.SyncResult
232233
if cmd.Conversation != "" {
233234
result, err = s.SyncConversation(ctx, rt.Config.DBPath, cmd.Conversation)
235+
} else if cmd.Resume {
236+
result, err = s.ResumeTail(ctx, rt.Config.DBPath, cmd.Limit)
234237
} else {
235238
updatedAfter, err := parseSince(cmd.UpdatedSince, time.Now().UTC())
236239
if err != nil {
@@ -248,18 +251,18 @@ func (cmd syncCmd) Run(ctx commandContext) error {
248251

249252
func (cmd syncCmd) validateMode() error {
250253
modes := 0
251-
for _, enabled := range []bool{cmd.Fixture != "", cmd.UpdatedSince != "", cmd.Conversation != ""} {
254+
for _, enabled := range []bool{cmd.Fixture != "", cmd.UpdatedSince != "", cmd.Conversation != "", cmd.Resume} {
252255
if enabled {
253256
modes++
254257
}
255258
}
256259
if modes == 0 {
257-
return output.UsageError{Err: fmt.Errorf("sync requires --fixture, --updated-since, or --conversation")}
260+
return output.UsageError{Err: fmt.Errorf("sync requires --fixture, --updated-since, --conversation, or --resume")}
258261
}
259262
if modes > 1 {
260-
return output.UsageError{Err: fmt.Errorf("sync accepts exactly one of --fixture, --updated-since, or --conversation")}
263+
return output.UsageError{Err: fmt.Errorf("sync accepts exactly one of --fixture, --updated-since, --conversation, or --resume")}
261264
}
262-
if cmd.UpdatedSince != "" && cmd.Limit < 0 {
265+
if (cmd.UpdatedSince != "" || cmd.Resume) && cmd.Limit < 0 {
263266
return output.UsageError{Err: fmt.Errorf("--limit must be >= 0")}
264267
}
265268
return nil

‎internal/cli/cli_test.go‎

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,10 +3,12 @@ package cli
33
import (
44
"bytes"
55
"context"
6+
"strings"
67
"testing"
78
"time"
89

910
"github.com/openclaw/crawlkit/output"
11+
"github.com/uinaf/fincrawl/internal/config"
1012
)
1113

1214
func TestSyncRejectsAmbiguousModes(t *testing.T) {
@@ -59,6 +61,39 @@ func TestSyncRejectsNegativeUpdatedSinceLimit(t *testing.T) {
5961
}
6062
}
6163

64+
func TestSyncRejectsNegativeResumeLimit(t *testing.T) {
65+
t.Setenv("FINCRAWL_HOME", t.TempDir())
66+
var stdout bytes.Buffer
67+
var stderr bytes.Buffer
68+
69+
err := Run(context.Background(), []string{
70+
"sync",
71+
"--resume",
72+
"--limit=-1",
73+
}, &stdout, &stderr)
74+
if !output.IsUsage(err) {
75+
t.Fatalf("expected usage error, got %v", err)
76+
}
77+
if stdout.Len() != 0 {
78+
t.Fatalf("stdout = %q", stdout.String())
79+
}
80+
}
81+
82+
func TestSyncResumeReachesLiveDispatch(t *testing.T) {
83+
t.Setenv("FINCRAWL_HOME", t.TempDir())
84+
t.Setenv(config.EnvIntercomCred, "fake-token")
85+
var stdout bytes.Buffer
86+
var stderr bytes.Buffer
87+
88+
err := Run(context.Background(), []string{"sync", "--resume", "--json"}, &stdout, &stderr)
89+
if err == nil || !strings.Contains(err.Error(), "no active Intercom tail sync state") {
90+
t.Fatalf("expected missing active state error, got %v", err)
91+
}
92+
if stdout.Len() != 0 {
93+
t.Fatalf("stdout = %q", stdout.String())
94+
}
95+
}
96+
6297
func TestVersionPrintsJSON(t *testing.T) {
6398
var stdout bytes.Buffer
6499
var stderr bytes.Buffer

‎internal/store/sync_state.go‎

Lines changed: 100 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,100 @@
1+
package store
2+
3+
import (
4+
"context"
5+
"database/sql"
6+
"fmt"
7+
"time"
8+
9+
ckstore "github.com/openclaw/crawlkit/store"
10+
)
11+
12+
const IntercomTailSyncStateID = "intercom.tail"
13+
14+
type SyncState struct {
15+
ID string `json:"id"`
16+
Provider string `json:"provider"`
17+
CursorKind string `json:"cursor_kind"`
18+
HighWaterMark string `json:"high_water_mark"`
19+
ActiveWindowStart string `json:"active_window_start"`
20+
ActiveWindowEnd string `json:"active_window_end"`
21+
LastProviderID string `json:"last_provider_id"`
22+
PageCursor string `json:"page_cursor"`
23+
UpdatedAt string `json:"updated_at"`
24+
}
25+
26+
func LoadSyncState(ctx context.Context, dbPath, id string) (SyncState, bool, error) {
27+
st, err := ckstore.Open(ctx, ckstore.Options{Path: dbPath, Schema: Schema, SchemaVersion: SchemaVersion})
28+
if err != nil {
29+
return SyncState{}, false, err
30+
}
31+
defer st.Close()
32+
var state SyncState
33+
err = st.DB().QueryRowContext(ctx, `select id, provider, cursor_kind, high_water_mark, active_window_start,
34+
active_window_end, last_provider_id, page_cursor, updated_at
35+
from sync_state where id = ?`, id).Scan(
36+
&state.ID,
37+
&state.Provider,
38+
&state.CursorKind,
39+
&state.HighWaterMark,
40+
&state.ActiveWindowStart,
41+
&state.ActiveWindowEnd,
42+
&state.LastProviderID,
43+
&state.PageCursor,
44+
&state.UpdatedAt,
45+
)
46+
if err == nil {
47+
return state, true, nil
48+
}
49+
if err == sql.ErrNoRows {
50+
return SyncState{}, false, nil
51+
}
52+
return SyncState{}, false, err
53+
}
54+
55+
func SaveSyncState(ctx context.Context, dbPath string, state SyncState) error {
56+
if state.ID == "" {
57+
return fmt.Errorf("sync state id is required")
58+
}
59+
if state.Provider == "" {
60+
state.Provider = ProviderIntercom
61+
}
62+
if state.CursorKind == "" {
63+
state.CursorKind = "updated_at"
64+
}
65+
state.UpdatedAt = time.Now().UTC().Format(time.RFC3339)
66+
st, err := ckstore.Open(ctx, ckstore.Options{Path: dbPath, Schema: Schema, SchemaVersion: SchemaVersion})
67+
if err != nil {
68+
return err
69+
}
70+
defer st.Close()
71+
return st.WithTx(ctx, func(tx *sql.Tx) error {
72+
_, err := tx.ExecContext(ctx, `insert into sync_state(
73+
id, provider, cursor_kind, high_water_mark, active_window_start,
74+
active_window_end, last_provider_id, page_cursor, updated_at
75+
) values(?, ?, ?, ?, ?, ?, ?, ?, ?)
76+
on conflict(id) do update set
77+
provider=excluded.provider,
78+
cursor_kind=excluded.cursor_kind,
79+
high_water_mark=excluded.high_water_mark,
80+
active_window_start=excluded.active_window_start,
81+
active_window_end=excluded.active_window_end,
82+
last_provider_id=excluded.last_provider_id,
83+
page_cursor=excluded.page_cursor,
84+
updated_at=excluded.updated_at`,
85+
state.ID,
86+
state.Provider,
87+
state.CursorKind,
88+
state.HighWaterMark,
89+
state.ActiveWindowStart,
90+
state.ActiveWindowEnd,
91+
state.LastProviderID,
92+
state.PageCursor,
93+
state.UpdatedAt,
94+
)
95+
if err != nil {
96+
return fmt.Errorf("save sync state: %w", err)
97+
}
98+
return nil
99+
})
100+
}

‎internal/store/sync_state_test.go‎

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
1+
package store
2+
3+
import (
4+
"context"
5+
"path/filepath"
6+
"testing"
7+
)
8+
9+
func TestSaveAndLoadSyncState(t *testing.T) {
10+
ctx := context.Background()
11+
dbPath := filepath.Join(t.TempDir(), "archive.db")
12+
state := SyncState{
13+
ID: IntercomTailSyncStateID,
14+
Provider: ProviderIntercom,
15+
CursorKind: "updated_at",
16+
HighWaterMark: "2026-05-16T10:00:00Z",
17+
ActiveWindowStart: "2026-05-16T09:00:00Z",
18+
ActiveWindowEnd: "2026-05-16T11:00:00Z",
19+
LastProviderID: "conv_fake_1",
20+
PageCursor: "cursor_fake_1",
21+
}
22+
if err := SaveSyncState(ctx, dbPath, state); err != nil {
23+
t.Fatal(err)
24+
}
25+
loaded, ok, err := LoadSyncState(ctx, dbPath, IntercomTailSyncStateID)
26+
if err != nil {
27+
t.Fatal(err)
28+
}
29+
if !ok {
30+
t.Fatalf("expected sync state")
31+
}
32+
if loaded.PageCursor != "cursor_fake_1" || loaded.LastProviderID != "conv_fake_1" {
33+
t.Fatalf("loaded state = %#v", loaded)
34+
}
35+
if loaded.UpdatedAt == "" {
36+
t.Fatalf("updated_at was not set")
37+
}
38+
}
39+
40+
func TestLoadSyncStateMissing(t *testing.T) {
41+
ctx := context.Background()
42+
dbPath := filepath.Join(t.TempDir(), "archive.db")
43+
_, ok, err := LoadSyncState(ctx, dbPath, IntercomTailSyncStateID)
44+
if err != nil {
45+
t.Fatal(err)
46+
}
47+
if ok {
48+
t.Fatalf("unexpected sync state")
49+
}
50+
}

0 commit comments

Comments
 (0)