History

feat(cloud): upload sqlite archive objects (#61)

Vincent Koc committed 5 months ago

Commit 7a33ea31 parent 0908f8b5 files · +405 −12Browse files

Showing the changes to internal/cli (2 of 5 files). Show all

Unified · Split

2 files
  1. internal/cli/app_test.go +258 −0
  2. internal/cli/cloud_commands.go +141 −9
internal/cli/app_test.go+258−0
@@ -3,9 +3,11 @@ package cli · 2 unmodified lines
33 import (
44 "bytes"
55 "context"
6+ "database/sql"
67 "encoding/json"
78 "errors"
89 "fmt"
10+ "io"
911 "net/http"
1012 "net/http/httptest"
1113 "os"
@@ -53,6 +55,231 @@ func TestInitWritesConfig(t *testing.T) { · 41 unmodified lines
5355 }
5456 }
5557
58+func TestGitcrawlThreadBodyExpr(t *testing.T) {
59+ ctx := context.Background()
60+ cases := []struct {
61+ name string
62+ schema string
63+ want string
64+ }{
65+ {name: "body and excerpt", schema: "create table threads(id text primary key, body text, body_excerpt text)", want: "coalesce(body, body_excerpt, '')"},
66+ {name: "body only", schema: "create table threads(id text primary key, body text)", want: "coalesce(body, '')"},
67+ {name: "excerpt only", schema: "create table threads(id text primary key, body_excerpt text)", want: "coalesce(body_excerpt, '')"},
68+ {name: "neither", schema: "create table threads(id text primary key)", want: "''"},
69+ }
70+ for _, tc := range cases {
71+ t.Run(tc.name, func(t *testing.T) {
72+ db, err := sql.Open("sqlite", filepath.Join(t.TempDir(), "threads.db"))
73+ if err != nil {
74+ t.Fatalf("open sqlite: %v", err)
75+ }
76+ defer db.Close()
77+ if _, err := db.ExecContext(ctx, tc.schema); err != nil {
78+ t.Fatalf("create schema: %v", err)
79+ }
80+ got, err := gitcrawlThreadBodyExpr(ctx, db)
81+ if err != nil {
82+ t.Fatalf("body expr: %v", err)
83+ }
84+ if got != tc.want {
85+ t.Fatalf("body expr = %q, want %q", got, tc.want)
86+ }
87+ })
88+ }
89+}
90+
91+func TestCloudSQLiteSnapshotFallbacks(t *testing.T) {
92+ ctx := context.Background()
93+ db, err := sql.Open("sqlite", filepath.Join(t.TempDir(), "closed.db"))
94+ if err != nil {
95+ t.Fatalf("open sqlite: %v", err)
96+ }
97+ if err := db.Close(); err != nil {
98+ t.Fatalf("close sqlite: %v", err)
99+ }
100+ fallback := filepath.Join(t.TempDir(), "fallback.db")
101+ got, cleanup, err := sqliteSnapshotPath(ctx, db, fallback)
102+ if err != nil {
103+ t.Fatalf("snapshot fallback: %v", err)
104+ }
105+ defer cleanup()
106+ if got != fallback {
107+ t.Fatalf("snapshot fallback path = %q, want %q", got, fallback)
108+ }
109+ if _, cleanup, err := sqliteSnapshotPath(ctx, db, ""); err == nil {
110+ cleanup()
111+ t.Fatal("snapshot without fallback unexpectedly succeeded")
112+ }
113+}
114+
115+func TestCloudSQLiteSnapshotVacuum(t *testing.T) {
116+ ctx := context.Background()
117+ db, err := sql.Open("sqlite", filepath.Join(t.TempDir(), "source.db"))
118+ if err != nil {
119+ t.Fatalf("open sqlite: %v", err)
120+ }
121+ defer db.Close()
122+ if _, err := db.ExecContext(ctx, "create table repositories(id text primary key)"); err != nil {
123+ t.Fatalf("create schema: %v", err)
124+ }
125+ if _, err := db.ExecContext(ctx, "insert into repositories(id) values('r1')"); err != nil {
126+ t.Fatalf("seed schema: %v", err)
127+ }
128+ snapshot, cleanup, err := sqliteSnapshotPath(ctx, db, "")
129+ if err != nil {
130+ t.Fatalf("snapshot: %v", err)
131+ }
132+ defer cleanup()
133+ snapshotDB, err := sql.Open("sqlite", snapshot)
134+ if err != nil {
135+ t.Fatalf("open snapshot: %v", err)
136+ }
137+ defer snapshotDB.Close()
138+ var count int
139+ if err := snapshotDB.QueryRowContext(ctx, "select count(*) from repositories").Scan(&count); err != nil {
140+ t.Fatalf("query snapshot: %v", err)
141+ }
142+ if count != 1 {
143+ t.Fatalf("snapshot count = %d, want 1", count)
144+ }
145+}
146+
147+func TestCloudFileSHA256(t *testing.T) {
148+ path := filepath.Join(t.TempDir(), "payload.txt")
149+ if err := os.WriteFile(path, []byte("abc"), 0o600); err != nil {
150+ t.Fatalf("write payload: %v", err)
151+ }
152+ got, err := cloudFileSHA256(path)
153+ if err != nil {
154+ t.Fatalf("hash payload: %v", err)
155+ }
156+ const want = "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad"
157+ if got != want {
158+ t.Fatalf("hash = %q, want %q", got, want)
159+ }
160+ if _, err := cloudFileSHA256(filepath.Join(t.TempDir(), "missing")); err == nil {
161+ t.Fatal("missing payload unexpectedly hashed")
162+ }
163+}
164+
165+func TestSendIngestRowsBatchesAndAcceptsEmpty(t *testing.T) {
166+ ctx := context.Background()
167+ var cursors []string
168+ var finals []bool
169+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
170+ if req.URL.EscapedPath() != "/v1/apps/gitcrawl/archives/gitcrawl%2Fopenclaw/ingest" {
171+ http.NotFound(w, req)
172+ return
173+ }
174+ var body crawlremote.IngestRequest
175+ if err := json.NewDecoder(req.Body).Decode(&body); err != nil {
176+ http.Error(w, err.Error(), http.StatusBadRequest)
177+ return
178+ }
179+ cursors = append(cursors, body.Cursor)
180+ finals = append(finals, body.Final)
181+ w.Header().Set("Content-Type", "application/json")
182+ _ = json.NewEncoder(w).Encode(crawlremote.IngestResult{Table: body.Table, RowsAccepted: int64(len(body.Rows)), Complete: body.Final})
183+ }))
184+ defer server.Close()
185+ client, err := crawlremote.NewClient(crawlremote.Options{
186+ Endpoint: server.URL,
187+ TokenProvider: crawlremote.StaticToken("token"),
188+ })
189+ if err != nil {
190+ t.Fatalf("remote client: %v", err)
191+ }
192+ manifest := crawlremote.IngestManifest{App: "gitcrawl", Archive: "gitcrawl/openclaw"}
193+ empty, err := sendIngestRows(ctx, client, "gitcrawl", "gitcrawl/openclaw", manifest, "repositories", gitcrawlRepositoryColumns, nil, false)
194+ if err != nil {
195+ t.Fatalf("empty ingest: %v", err)
196+ }
197+ if empty != 0 {
198+ t.Fatalf("empty rows accepted = %d, want 0", empty)
199+ }
200+ rows := make([][]any, gitcrawlCloudBatchSize+1)
201+ for i := range rows {
202+ rows[i] = []any{fmt.Sprintf("r%d", i), "openclaw/repo", "openclaw", "repo", "", "", ""}
203+ }
204+ total, err := sendIngestRows(ctx, client, "gitcrawl", "gitcrawl/openclaw", manifest, "repositories", gitcrawlRepositoryColumns, rows, true)
205+ if err != nil {
206+ t.Fatalf("batched ingest: %v", err)
207+ }
208+ if total != int64(len(rows)) {
209+ t.Fatalf("rows accepted = %d, want %d", total, len(rows))
210+ }
211+ if got, want := strings.Join(cursors, ","), ",,250"; got != want {
212+ t.Fatalf("cursors = %q, want %q", got, want)
213+ }
214+ if got, want := fmt.Sprint(finals), "[false false true]"; got != want {
215+ t.Fatalf("finals = %s, want %s", got, want)
216+ }
217+}
218+
219+func TestRemoteValueCoercionHelpers(t *testing.T) {
220+ value := map[string]any{
221+ "bool_string": "true",
222+ "bool_number": json.Number("1"),
223+ "int_string": "42",
224+ "int_number": json.Number("43"),
225+ "int64_string": "42000000000",
226+ "float_string": "4.25",
227+ "float_number": json.Number("5.5"),
228+ }
229+ if !boolValue(value, "bool_string") || !boolValue(value, "bool_number") || boolValue(value, "missing") {
230+ t.Fatalf("bool coercion failed")
231+ }
232+ if got := intValue(value, "int_string"); got != 42 {
233+ t.Fatalf("int string = %d, want 42", got)
234+ }
235+ if got := intValue(value, "int_number"); got != 43 {
236+ t.Fatalf("int number = %d, want 43", got)
237+ }
238+ if got := int64Value(value, "int64_string"); got != 42000000000 {
239+ t.Fatalf("int64 string = %d, want 42000000000", got)
240+ }
241+ if got := floatValue(value, "float_string"); got != 4.25 {
242+ t.Fatalf("float string = %v, want 4.25", got)
243+ }
244+ if got := floatValue(value, "float_number"); got != 5.5 {
245+ t.Fatalf("float number = %v, want 5.5", got)
246+ }
247+ ref := threadReference{Owner: "openclaw", Repo: "gitcrawl"}
248+ if got := ref.FullName(); got != "openclaw/gitcrawl" {
249+ t.Fatalf("full name = %q, want openclaw/gitcrawl", got)
250+ }
251+ if got := (threadReference{Owner: "openclaw"}).FullName(); got != "" {
252+ t.Fatalf("partial full name = %q, want empty", got)
253+ }
254+}
255+
256+func TestUploadSQLiteArchiveReturnsRemoteError(t *testing.T) {
257+ ctx := context.Background()
258+ db, err := sql.Open("sqlite", filepath.Join(t.TempDir(), "source.db"))
259+ if err != nil {
260+ t.Fatalf("open sqlite: %v", err)
261+ }
262+ defer db.Close()
263+ if _, err := db.ExecContext(ctx, "create table repositories(id text primary key)"); err != nil {
264+ t.Fatalf("create schema: %v", err)
265+ }
266+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
267+ http.Error(w, "nope", http.StatusBadGateway)
268+ }))
269+ defer server.Close()
270+ client, err := crawlremote.NewClient(crawlremote.Options{
271+ Endpoint: server.URL,
272+ TokenProvider: crawlremote.StaticToken("token"),
273+ })
274+ if err != nil {
275+ t.Fatalf("remote client: %v", err)
276+ }
277+ _, err = uploadSQLiteArchive(ctx, client, "gitcrawl", "gitcrawl/openclaw", db, "", crawlremote.IngestManifest{SchemaName: "gitcrawl-cloud-v1"})
278+ if err == nil {
279+ t.Fatal("upload unexpectedly succeeded")
280+ }
281+}
282+
56283 func TestInitDefaultOutputIsHumanReadable(t *testing.T) {
57284 dir := t.TempDir()
58285 configPath := filepath.Join(dir, "config.toml")
@@ -227,11 +454,36 @@ func TestCloudPublishSendsLocalRows(t *testing.T) { · 168 unmodified lines
227454 tokenEnv := "GITCRAWL_TEST_PUBLISH_TOKEN"
228455 t.Setenv(tokenEnv, "publish-token")
229456 seenTables := map[string]crawlremote.IngestRequest{}
457+ var sawSQLiteUpload bool
230458 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
231459 if got := r.Header.Get("authorization"); got != "Bearer publish-token" {
232460 http.Error(w, "missing bearer", http.StatusUnauthorized)
233461 return
234462 }
463+ if r.Method == http.MethodPut && r.URL.EscapedPath() == "/v1/apps/gitcrawl/archives/gitcrawl%2Fopenclaw__openclaw/sqlite" {
464+ sawSQLiteUpload = true
465+ payload, err := io.ReadAll(r.Body)
466+ if err != nil {
467+ http.Error(w, err.Error(), http.StatusBadRequest)
468+ return
469+ }
470+ if !bytes.HasPrefix(payload, []byte("SQLite format 3")) {
471+ http.Error(w, "not sqlite", http.StatusBadRequest)
472+ return
473+ }
474+ if r.Header.Get("x-crawl-content-sha256") == "" {
475+ http.Error(w, "missing hash", http.StatusBadRequest)
476+ return
477+ }
478+ w.Header().Set("content-type", "application/json")
479+ _ = json.NewEncoder(w).Encode(crawlremote.SQLiteUploadResult{
480+ App: "gitcrawl",
481+ Archive: "gitcrawl/openclaw__openclaw",
482+ Complete: true,
483+ Object: &crawlremote.SQLiteObject{Key: "v1/gitcrawl/gitcrawl%2Fopenclaw__openclaw/sqlite/current.db", Size: int64(len(payload))},
484+ })
485+ return
486+ }
235487 if r.Method != http.MethodPost || r.URL.EscapedPath() != "/v1/apps/gitcrawl/archives/gitcrawl%2Fopenclaw__openclaw/ingest" {
236488 http.NotFound(w, r)
237489 return
@@ -277,6 +529,9 @@ func TestCloudPublishSendsLocalRows(t *testing.T) { · 39 unmodified lines
277529 if !seenTables["threads"].Final {
278530 t.Fatalf("threads batch should finish ingest")
279531 }
532+ if !sawSQLiteUpload {
533+ t.Fatalf("cloud publish did not upload sqlite object")
534+ }
280535 var payload map[string]any
281536 if err := json.Unmarshal(out.Bytes(), &payload); err != nil {
282537 t.Fatalf("decode output: %v\n%s", err, out.String())
@@ -284,6 +539,9 @@ func TestCloudPublishSendsLocalRows(t *testing.T) { · 1 unmodified line
284539 if payload["repositories"] != float64(1) || payload["threads"] != float64(3) {
285540 t.Fatalf("unexpected output: %#v", payload)
286541 }
542+ if payload["sqlite_object"] == nil {
543+ t.Fatalf("missing sqlite object output: %#v", payload)
544+ }
287545 }
288546
289547 func TestRemoteLoginStoresKeyringToken(t *testing.T) {
internal/cli/cloud_commands.go+141−9
@@ -2,11 +2,16 @@ package cli · 1 unmodified line
22
33 import (
44 "context"
5+ "crypto/sha256"
56 "database/sql"
67 "flag"
78 "fmt"
89 "io"
10+ "net/http"
11+ "os"
12+ "path/filepath"
913 "strings"
14+ "time"
1015
1116 crawlremote "github.com/openclaw/crawlkit/remote"
1217 "github.com/openclaw/gitcrawl/internal/config"
@@ -67,7 +72,8 @@ func (a *App) runCloudPublish(ctx context.Context, args []string) error { · 54 unmodified lines
6772 TokenEnv: firstNonEmpty(*tokenEnv, cfg.Remote.TokenEnv, crawlremote.DefaultTokenEnv),
6873 }
6974 client, err := crawlremote.NewClientFromConfig(remoteCfg, crawlremote.Options{
70− UserAgent: "gitcrawl/" + version,
75+ UserAgent: "gitcrawl/" + version,
76+ HTTPClient: &http.Client{Timeout: 10 * time.Minute},
7177 })
7278 if err != nil {
7379 return err
@@ -85,7 +91,11 @@ func (a *App) runCloudPublish(ctx context.Context, args []string) error { · 11 unmodified lines
8591 if err != nil {
8692 return err
8793 }
88− threadRows, err := publishRows(ctx, rt.Store.DB(), gitcrawlThreadExportSQL, func(values []any) []any { return values })
94+ threadSQL, err := gitcrawlThreadExportSQL(ctx, rt.Store.DB())
95+ if err != nil {
96+ return err
97+ }
98+ threadRows, err := publishRows(ctx, rt.Store.DB(), threadSQL, func(values []any) []any { return values })
8999 if err != nil {
90100 return err
91101 }
@@ -97,11 +107,16 @@ func (a *App) runCloudPublish(ctx context.Context, args []string) error { · 5 unmodified lines
97107 if err != nil {
98108 return err
99109 }
110+ sqliteObject, err := uploadSQLiteArchive(ctx, client, "gitcrawl", archiveID, rt.Store.DB(), rt.SourceDBPath, manifest)
111+ if err != nil {
112+ return err
113+ }
100114 return a.writeOutput("cloud publish", map[string]any{
101− "remote": strings.TrimRight(endpoint, "/"),
102− "archive": archiveID,
103− "repositories": repoAccepted,
104− "threads": threadAccepted,
115+ "remote": strings.TrimRight(endpoint, "/"),
116+ "archive": archiveID,
117+ "repositories": repoAccepted,
118+ "threads": threadAccepted,
119+ "sqlite_object": sqliteObject,
105120 }, true)
106121 }
107122
@@ -175,6 +190,74 @@ func cursorFor(start int) string { · 67 unmodified lines
175190 return fmt.Sprintf("%d", start)
176191 }
177192
193+func uploadSQLiteArchive(ctx context.Context, client *crawlremote.Client, app, archive string, db *sql.DB, dbPath string, manifest crawlremote.IngestManifest) (*crawlremote.SQLiteObject, error) {
194+ snapshotPath, cleanup, err := sqliteSnapshotPath(ctx, db, dbPath)
195+ if err != nil {
196+ return nil, err
197+ }
198+ defer cleanup()
199+ file, err := os.Open(snapshotPath)
200+ if err != nil {
201+ return nil, fmt.Errorf("open sqlite snapshot: %w", err)
202+ }
203+ defer file.Close()
204+ info, err := file.Stat()
205+ if err != nil {
206+ return nil, fmt.Errorf("stat sqlite snapshot: %w", err)
207+ }
208+ sum, err := cloudFileSHA256(snapshotPath)
209+ if err != nil {
210+ return nil, err
211+ }
212+ if _, err := file.Seek(0, io.SeekStart); err != nil {
213+ return nil, fmt.Errorf("rewind sqlite snapshot: %w", err)
214+ }
215+ result, err := client.UploadSQLite(ctx, app, archive, crawlremote.SQLiteUploadRequest{
216+ Body: file,
217+ Size: info.Size(),
218+ ContentSHA256: sum,
219+ SchemaName: manifest.SchemaName,
220+ SchemaVersion: manifest.SchemaVersion,
221+ SchemaHash: manifest.SchemaHash,
222+ SourceSyncAt: manifest.SourceSyncAt,
223+ })
224+ if err != nil {
225+ return nil, err
226+ }
227+ return result.Object, nil
228+}
229+
230+func sqliteSnapshotPath(ctx context.Context, db *sql.DB, dbPath string) (string, func(), error) {
231+ tmpDir, err := os.MkdirTemp("", "gitcrawl-cloud-sqlite-*")
232+ if err != nil {
233+ return "", func() {}, fmt.Errorf("create sqlite snapshot dir: %w", err)
234+ }
235+ cleanup := func() { _ = os.RemoveAll(tmpDir) }
236+ snapshotPath := filepath.Join(tmpDir, "archive.db")
237+ if _, err := db.ExecContext(ctx, "vacuum main into ?", snapshotPath); err == nil {
238+ return snapshotPath, cleanup, nil
239+ }
240+ source := strings.TrimSpace(dbPath)
241+ if source == "" {
242+ cleanup()
243+ return "", func() {}, fmt.Errorf("sqlite snapshot failed and no source db path is available")
244+ }
245+ return source, cleanup, nil
246+}
247+
248+func cloudFileSHA256(path string) (string, error) {
249+ file, err := os.Open(path)
250+ if err != nil {
251+ return "", fmt.Errorf("open sqlite snapshot for hash: %w", err)
252+ }
253+ defer file.Close()
254+ hash := sha256.New()
255+ if _, err := io.Copy(hash, file); err != nil {
256+ return "", fmt.Errorf("hash sqlite snapshot: %w", err)
257+ }
258+ return fmt.Sprintf("%x", hash.Sum(nil)), nil
259+}
260+
178261 var gitcrawlRepositoryColumns = []string{"id", "full_name", "owner", "name", "html_url", "default_branch", "updated_at"}
179262
180263 const gitcrawlRepositoryExportSQL = `
@@ -184,11 +267,60 @@ order by id` · 3 unmodified lines
184267
185268 var gitcrawlThreadColumns = []string{"id", "repo_id", "github_id", "number", "kind", "state", "title", "body", "author_login", "author_type", "html_url", "labels_json", "assignees_json", "is_draft", "created_at_gh", "updated_at_gh", "closed_at_gh", "merged_at_gh", "updated_at"}
186269
187−const gitcrawlThreadExportSQL = `
188−select id, repo_id, github_id, number, kind, state, title, coalesce(body, ''),
270+func gitcrawlThreadExportSQL(ctx context.Context, db *sql.DB) (string, error) {
271+ bodyExpr, err := gitcrawlThreadBodyExpr(ctx, db)
272+ if err != nil {
273+ return "", err
274+ }
275+ return fmt.Sprintf(`
276+select id, repo_id, github_id, number, kind, state, title, %s,
189277 coalesce(author_login, ''), coalesce(author_type, ''), html_url,
190278 labels_json, assignees_json, is_draft, coalesce(created_at_gh, ''),
191279 coalesce(updated_at_gh, ''), coalesce(closed_at_gh, ''),
192280 coalesce(merged_at_gh, ''), updated_at
193281 from threads
194−order by repo_id, number`
282+order by repo_id, number`, bodyExpr), nil
283+}
284+
285+func gitcrawlThreadBodyExpr(ctx context.Context, db *sql.DB) (string, error) {
286+ hasBody, err := sqliteColumnExists(ctx, db, "threads", "body")
287+ if err != nil {
288+ return "", err
289+ }
290+ hasExcerpt, err := sqliteColumnExists(ctx, db, "threads", "body_excerpt")
291+ if err != nil {
292+ return "", err
293+ }
294+ switch {
295+ case hasBody && hasExcerpt:
296+ return "coalesce(body, body_excerpt, '')", nil
297+ case hasBody:
298+ return "coalesce(body, '')", nil
299+ case hasExcerpt:
300+ return "coalesce(body_excerpt, '')", nil
301+ default:
302+ return "''", nil
303+ }
304+}
305+
306+func sqliteColumnExists(ctx context.Context, db *sql.DB, table, column string) (bool, error) {
307+ rows, err := db.QueryContext(ctx, `pragma table_info(`+table+`)`)
308+ if err != nil {
309+ return false, err
310+ }
311+ defer rows.Close()
312+ for rows.Next() {
313+ var cid int
314+ var name, typ string
315+ var notNull int
316+ var defaultValue any
317+ var pk int
318+ if err := rows.Scan(&cid, &name, &typ, &notNull, &defaultValue, &pk); err != nil {
319+ return false, err
320+ }
321+ if name == column {
322+ return true, nil
323+ }
324+ }
325+ return false, rows.Err()
326+}