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

Author: vincentkoc

CHANGELOG.md

@@ -4,7 +4,7 @@
 
 ### Changes
 
-- Update `crawlkit` to v0.8.0.
+- Update `crawlkit` to v0.10.0.
 - Add read-only Cloudflare remote archive scaffolding with `[remote]` config,
   `subscribe-cloud`, GitHub-backed `remote login` with OAuth or token-env
   bootstrap, `remote status`, `remote archives`, and cloud-mode `status --json`
@@ -14,6 +14,9 @@
 - Add `discrawl cloud publish` to export non-DM local SQLite rows into the
   Cloudflare remote archive ingest API without changing Git snapshot
   publishing.
+- Mirror the non-DM local SQLite archive into the Worker-backed R2 object store
+  during `discrawl cloud publish`, alongside the D1 row ingest used for live
+  queries.
 
 ### Fixes
 

go.mod

@@ -44,7 +44,7 @@ require (
 	github.com/muesli/cancelreader v0.2.2 // indirect
 	github.com/muesli/termenv v0.16.0 // indirect
 	github.com/ncruces/go-strftime v1.0.0 // indirect
-	github.com/openclaw/crawlkit v0.8.0
+	github.com/openclaw/crawlkit v0.10.0
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
 	github.com/rivo/uniseg v0.4.7 // indirect

go.sum

@@ -71,8 +71,8 @@ github.com/muesli/termenv v0.16.0 h1:S5AlUN9dENB57rsbnkPyfdGuWIlkmzJjbFf0Tf5FWUc
 github.com/muesli/termenv v0.16.0/go.mod h1:ZRfOIKPFDYQoDFF4Olj7/QJbW60Ol/kL1pU3VfY/Cnk=
 github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w=
 github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls=
-github.com/openclaw/crawlkit v0.8.0 h1:EZEAMy7dI6zVn+AC6lN8GycMY8bV3TF3KnheG55A0uU=
-github.com/openclaw/crawlkit v0.8.0/go.mod h1:2XkSx3N8yzyjc+Jyf1Zl9VEQVlElh/dzBMW1cRMQQGw=
+github.com/openclaw/crawlkit v0.10.0 h1:FKe5Aus+lgik76KswimTvnlwMdfiuaZV1MGign6z73k=
+github.com/openclaw/crawlkit v0.10.0/go.mod h1:2XkSx3N8yzyjc+Jyf1Zl9VEQVlElh/dzBMW1cRMQQGw=
 github.com/pelletier/go-toml/v2 v2.3.1 h1:MYEvvGnQjeNkRF1qUuGolNtNExTDwct51yp7olPtrEc=
 github.com/pelletier/go-toml/v2 v2.3.1/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY=
 github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA=

internal/cli/cli_test.go

@@ -3,6 +3,7 @@ package cli
 import (
 	"bytes"
 	"context"
+	"database/sql"
 	"encoding/json"
 	"errors"
 	"flag"
@@ -92,6 +93,10 @@ func TestCommandValidationEdges(t *testing.T) {
 		{"--config", cfgPath, "sync", "--source", "bogus"},
 		{"--config", cfgPath, "sync", "--since", "not-time"},
 		{"--config", cfgPath, "sync", "--no-update", "--update", "force"},
+		{"--config", cfgPath, "cloud", "publish", "--bogus"},
+		{"--config", cfgPath, "cloud", "publish", "extra"},
+		{"--config", cfgPath, "cloud", "publish", "--json"},
+		{"--config", cfgPath, "cloud", "publish", "--remote", "https://remote.example"},
 		{"--config", cfgPath, "publish", "--remote", ""},
 		{"--config", cfgPath, "subscribe"},
 		{"--config", cfgPath, "update", "extra"},
@@ -1300,8 +1305,59 @@ func TestCloudPublishSendsNonDMRows(t *testing.T) {
 	tokenEnv := "DISCRAWL_TEST_PUBLISH_TOKEN"
 	t.Setenv(tokenEnv, "publish-token")
 	seenTables := map[string]crawlremote.IngestRequest{}
+	var sawSQLiteUpload bool
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
 		assert.Equal(t, "Bearer publish-token", req.Header.Get("Authorization"))
+		if req.Method == http.MethodPut && req.URL.EscapedPath() == "/v1/apps/discrawl/archives/discrawl%2Fopenclaw/sqlite" {
+			sawSQLiteUpload = true
+			payload, err := io.ReadAll(req.Body)
+			if err != nil {
+				t.Errorf("read sqlite upload: %v", err)
+				http.Error(w, err.Error(), http.StatusBadRequest)
+				return
+			}
+			if !bytes.HasPrefix(payload, []byte("SQLite format 3")) {
+				t.Errorf("sqlite upload should contain a sqlite image")
+				http.Error(w, "not sqlite", http.StatusBadRequest)
+				return
+			}
+			uploadPath := filepath.Join(dir, "uploaded-cloud.db")
+			if err := os.WriteFile(uploadPath, payload, 0o600); err != nil {
+				t.Errorf("write sqlite upload: %v", err)
+				http.Error(w, err.Error(), http.StatusInternalServerError)
+				return
+			}
+			uploadDB, err := sql.Open("sqlite", uploadPath)
+			if err != nil {
+				t.Errorf("open sqlite upload: %v", err)
+				http.Error(w, err.Error(), http.StatusInternalServerError)
+				return
+			}
+			defer func() { _ = uploadDB.Close() }()
+			var dmMessages int
+			if err := uploadDB.QueryRowContext(ctx, "select count(*) from messages where guild_id = ?", store.DirectMessageGuildID).Scan(&dmMessages); err != nil {
+				t.Errorf("count dm messages: %v", err)
+				http.Error(w, err.Error(), http.StatusInternalServerError)
+				return
+			}
+			assert.Zero(t, dmMessages)
+			var tableCount int
+			if err := uploadDB.QueryRowContext(ctx, "select count(*) from sqlite_master where type = 'table' and name in ('guilds', 'channels', 'members', 'messages')").Scan(&tableCount); err != nil {
+				t.Errorf("count cloud tables: %v", err)
+				http.Error(w, err.Error(), http.StatusInternalServerError)
+				return
+			}
+			assert.Equal(t, 4, tableCount)
+			assert.NotEmpty(t, req.Header.Get("X-Crawl-Content-Sha256"))
+			w.Header().Set("Content-Type", "application/json")
+			_ = json.NewEncoder(w).Encode(crawlremote.SQLiteUploadResult{
+				App:      "discrawl",
+				Archive:  "discrawl/openclaw",
+				Complete: true,
+				Object:   &crawlremote.SQLiteObject{Key: "v1/discrawl/discrawl%2Fopenclaw/sqlite/current.db", Size: int64(len(payload))},
+			})
+			return
+		}
 		assert.Equal(t, http.MethodPost, req.Method)
 		assert.Equal(t, "/v1/apps/discrawl/archives/discrawl%2Fopenclaw/ingest", req.URL.EscapedPath())
 		var body crawlremote.IngestRequest
@@ -1339,11 +1395,101 @@ func TestCloudPublishSendsNonDMRows(t *testing.T) {
 	require.Len(t, seenTables["channels"].Rows, 1)
 	require.Len(t, seenTables["messages"].Rows, 1)
 	require.True(t, seenTables["messages"].Final)
+	require.True(t, sawSQLiteUpload)
 
 	var payload map[string]any
 	require.NoError(t, json.Unmarshal(out.Bytes(), &payload))
 	require.InDelta(t, float64(1), payload["guilds"], 0)
 	require.InDelta(t, float64(1), payload["messages"], 0)
+	require.NotNil(t, payload["sqlite_object"])
+}
+
+func TestCloudSQLiteExportHelpers(t *testing.T) {
+	ctx := context.Background()
+	dir := t.TempDir()
+	sourcePath := filepath.Join(dir, "source.db")
+	s := seedCLIStore(t, sourcePath)
+	require.NoError(t, s.UpsertMember(ctx, store.MemberRecord{
+		GuildID:     "g1",
+		UserID:      "u1",
+		Username:    "peter",
+		DisplayName: "Peter",
+		RoleIDsJSON: `[]`,
+		RawJSON:     `{"private":"ignored"}`,
+	}))
+	require.NoError(t, addCLIDMAttachment(ctx, s))
+
+	require.Empty(t, sqlPlaceholders(0))
+	require.Equal(t, "?,?,?", sqlPlaceholders(3))
+
+	snapshotPath, cleanup, err := sqliteSnapshotPath(ctx, s.DB())
+	require.NoError(t, err)
+	require.FileExists(t, snapshotPath)
+	cleanup()
+	require.NoFileExists(t, snapshotPath)
+
+	exportPath := filepath.Join(dir, "cloud.db")
+	require.NoError(t, writeCloudSQLiteExport(ctx, s.DB(), exportPath))
+	require.NoError(t, s.Close())
+
+	sum, err := cloudFileSHA256(exportPath)
+	require.NoError(t, err)
+	require.Len(t, sum, 64)
+	_, err = cloudFileSHA256(filepath.Join(dir, "missing.db"))
+	require.Error(t, err)
+
+	cloudDB, err := sql.Open("sqlite", exportPath)
+	require.NoError(t, err)
+	defer func() { _ = cloudDB.Close() }()
+
+	var guilds, channels, members, messages int
+	require.NoError(t, cloudDB.QueryRowContext(ctx, "select count(*) from guilds").Scan(&guilds))
+	require.NoError(t, cloudDB.QueryRowContext(ctx, "select count(*) from channels").Scan(&channels))
+	require.NoError(t, cloudDB.QueryRowContext(ctx, "select count(*) from members").Scan(&members))
+	require.NoError(t, cloudDB.QueryRowContext(ctx, "select count(*) from messages").Scan(&messages))
+	require.Equal(t, 1, guilds)
+	require.Equal(t, 1, channels)
+	require.Equal(t, 1, members)
+	require.Equal(t, 1, messages)
+
+	var dmRows int
+	require.NoError(t, cloudDB.QueryRowContext(ctx, "select count(*) from messages where guild_id = ?", store.DirectMessageGuildID).Scan(&dmRows))
+	require.Zero(t, dmRows)
+
+	var authorUsername string
+	require.NoError(t, cloudDB.QueryRowContext(ctx, "select author_username from messages where message_id = 'm100'").Scan(&authorUsername))
+	require.Equal(t, "Peter", authorUsername)
+
+	var rawJSONColumns int
+	require.NoError(t, cloudDB.QueryRowContext(ctx, "select count(*) from pragma_table_info('messages') where name = 'raw_json'").Scan(&rawJSONColumns))
+	require.Zero(t, rawJSONColumns)
+}
+
+func TestCopyCloudSQLiteRowsErrorsAndBytes(t *testing.T) {
+	ctx := context.Background()
+	dir := t.TempDir()
+	source, err := sql.Open("sqlite", filepath.Join(dir, "source.db"))
+	require.NoError(t, err)
+	defer func() { _ = source.Close() }()
+	out, err := sql.Open("sqlite", filepath.Join(dir, "out.db"))
+	require.NoError(t, err)
+	defer func() { _ = out.Close() }()
+
+	_, err = source.ExecContext(ctx, `create table source_rows(value blob)`)
+	require.NoError(t, err)
+	_, err = source.ExecContext(ctx, `insert into source_rows(value) values(x'68656c6c6f')`)
+	require.NoError(t, err)
+	_, err = out.ExecContext(ctx, `create table copied(value text)`)
+	require.NoError(t, err)
+
+	require.NoError(t, copyCloudSQLiteRows(ctx, source, out, "copied", []string{"value"}, `select value from source_rows`))
+	var value string
+	require.NoError(t, out.QueryRowContext(ctx, `select value from copied`).Scan(&value))
+	require.Equal(t, "hello", value)
+
+	err = copyCloudSQLiteRows(ctx, source, out, "copied", []string{"value"}, `select missing from source_rows`)
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "query sqlite cloud export copied")
 }
 
 func TestShareCommandsPublishSubscribeAndUpdate(t *testing.T) {