diff --git a/CHANGELOG.md b/CHANGELOG.md index 59d38d4..37dde88 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,16 @@ ## Unreleased +### Changes + +- Added generic external archive providers with an explicit `provider:` sync source, a local JSONL subprocess protocol, scoped resumable checkpoints, bounded validation imports, and source-priority safeguards. + +### Performance + +- Batched unchanged-message checks and aligned search-index row IDs during external archive replays to avoid per-message database round trips and full-index replacement scans. + +### Maintenance + - Standardized the Makefile's build, check, snapshot, and fail-closed release targets across the crawler repositories. - Refreshed terminal detection and Unicode display-width dependencies. - Updated CrawlKit to 0.14.4, SQLite to 1.55.0, `golang.org/x/net` to 0.57.0, and replaced the retracted libc 1.74.3 with 1.74.4. diff --git a/README.md b/README.md index 7a42a9c..986144c 100644 --- a/README.md +++ b/README.md @@ -19,13 +19,14 @@ Slack search is convenient until you need your own workflow, your own retention, or your own queries. `slacrawl` is a Go-based CLI that pulls Slack workspace metadata and message history into SQLite so you can inspect it without depending on the Slack UI. -Data stays on your machine. You can run it in API mode, MCP connector mode, desktop mode, or a hybrid workflow. That covers one-shot syncs, live tailing over Socket Mode, connector-backed fetching, and local desktop recovery or "wiretap" style inspection from Slack Desktop artifacts already on your machine. +Data stays on your machine. You can run it in API mode, MCP connector mode, desktop mode, external-provider mode, or a hybrid workflow. That covers one-shot syncs, live tailing over Socket Mode, connector-backed fetching, local archive imports, and local desktop recovery or "wiretap" style inspection from Slack Desktop artifacts already on your machine. ## Included - local SQLite storage with full-text search backed by SQLite FTS5 - workspace, channel, user, and message sync - MCP connector sync through Codex's HTTP Slack connector or the reference Slack MCP server +- generic external archive providers over a local JSONL subprocess protocol - thread reply backfill when a user token is available - DM and MPIM sync when a user token is available - incremental API history sync by default, with `--full` reserved for deliberate backfills @@ -171,8 +172,9 @@ Treat one `slacrawl` config/database as one Slack visibility boundary. The archi - `--source bot` is an alias for `--source api`; it crawls Slack through configured bot/user tokens - `--source mcp` fetches through Codex's HTTP Slack connector or the reference Slack MCP server over stdio and requires a workspace ID for archive ownership +- `--source provider:` imports a configured external archive through a local subprocess and requires a workspace ID for archive ownership - `--source wiretap` is an alias for `--source desktop`; it reads the local Slack Desktop cache -- `--source all` runs API first, then desktop enrichment +- `--source all` runs API first, then desktop enrichment; external providers remain explicit - `[share]` is a backup/restore target for the current DB, not a second Slack source For separate company and personal archives, use separate configs with separate `db_path` and `[share].remote` values. @@ -183,6 +185,7 @@ Choose the path that matches your setup: - use `sync --source bot --full` only when you want a deliberate full backfill - use `sync --source bot --latest-only` when you only want fresh deltas on channels that already have local history - use `sync --source mcp --workspace T01234567` when a configured HTTP or stdio MCP connector can read the workspace +- use `sync --source provider:archive --workspace T01234567` when a configured local provider should import another archive - use `sync --source wiretap` when you want local desktop recovery only - use `watch` when you want desktop-local state to refresh into SQLite continuously @@ -194,7 +197,7 @@ Choose the path that matches your setup: - `publish` exports the local SQLite archive into a git repo as compressed JSONL shards plus a manifest - `subscribe` configures a git-backed reader that can run without Slack credentials - `update` safely merges the latest git snapshot; `update --restore` performs explicit exact replacement, including historical tag/ref restores -- `sync` performs a one-shot crawl from bot/API, MCP connector, wiretap/desktop, or both +- `sync` performs a one-shot crawl from bot/API, MCP connector, an external provider, wiretap/desktop, or the combined API-plus-desktop mode - `import` imports a Slack export ZIP or extracted export directory - `purge` previews or deletes messages and message-owned records older than a cutoff, with optional retained-event compaction - `tail` listens for live events through Socket Mode, including one tail per configured workspace @@ -362,6 +365,65 @@ go run ./cmd/slacrawl files fetch --missing --max-bytes 104857600 go run ./cmd/slacrawl sync --source bot --latest-only --with-media ``` +## External Archive Providers + +An external provider is a trusted local executable that streams workspace, +channel, user, and message records into the normal SQLite archive. Configure it +with a unique name, then select it explicitly with `provider:`: + +```toml +[[providers]] +name = "archive" +command = "/usr/local/bin/archive-provider" +args = ["provide", "--format", "jsonl"] +env_allowlist = ["ARCHIVE_DB_PATH"] +source_rank = 5 +batch_size = 1000 +``` + +```bash +slacrawl sync --source provider:archive --workspace T01234567 +slacrawl sync --source provider:archive --workspace T01234567 --latest-only +slacrawl sync --source provider:archive --workspace T01234567 --full +slacrawl sync --source provider:archive --workspace T01234567 --limit 100 +``` + +`command` must be an absolute path; `~` and `~/...` are expanded before +validation. The +configured `args` are passed directly without a shell. Only a minimal runtime +environment and variables named by `env_allowlist` reach the process, so put +secret values in the environment rather than in TOML. `source_rank` must be +greater than `2`; lower numeric ranks win, keeping higher-priority native data +authoritative. Equal ranks may replace, so use a larger number for a +lower-priority provider. + +`batch_size` defaults to `1000`, must be between `1` and `100000`, and is sent to +the provider as a preferred upstream batch size. + +`slacrawl` writes one JSON request line to provider stdin. The request identifies +protocol `slacrawl-provider-v1` and forwards the workspace, `since`, `full`, +`latest_only`, channel filters, opaque saved checkpoint, batch size, and optional +validation limit. Provider stdout must be JSONL: one `hello` record, zero or more +`workspace`, `channel`, `user`, `message`, or `checkpoint` records, then one +`done` record. A successful run must emit `done` and exit zero. `slacrawl` +validates workspace ownership and the message limit, builds normalized search +text, updates FTS and mentions, and does not replace a higher-priority canonical +message. + +Checkpoints are isolated by invocation scope. A normal incremental run reuses +the workspace checkpoint; filters, `--since`, `--full`, `--latest-only`, and +`--limit` use separate checkpoints so bounded or partial runs cannot advance the +normal cursor. A completed unbounded full run promotes its last checkpoint to +the matching incremental scope. Providers should emit a checkpoint only after +the records it covers and must honor the forwarded filters; `slacrawl` cannot +infer provider-specific filtering. + +Treat provider executables as part of the archive's trust boundary. They can +read allowlisted environment values, run with the caller's OS permissions, and +emit private Slack data. See +[Configuration](./docs/configuration.md#external-archive-providers) for the +complete request and response contract. + ## Git Archive Sharing Use git-share mode when one machine has Slack credentials and should publish snapshots, while other machines only need a local read-only archive. diff --git a/SPEC.md b/SPEC.md index 94c1497..98a45fc 100644 --- a/SPEC.md +++ b/SPEC.md @@ -26,6 +26,7 @@ V1 scope: - FTS5 search - raw SQL access - desktop-local Slack discovery on macOS and Linux +- external archive ingestion through a local JSONL provider protocol Out of scope for V1: @@ -43,7 +44,7 @@ Out of scope for V1: - language: Go - schema: single-workspace default, multi-workspace-ready - search: FTS5 first, embeddings later -- source precedence: user-token API, then bot-token API and slack-export imports, then desktop-local cache +- source precedence: user-token API, then bot-token API and slack-export imports, then desktop-local cache; external providers must use a numeric rank greater than `2`, and equal ranks may replace - files: metadata only in DB for V1 - future file-blob backup must store Git-share media as gzip-compressed files, import those files back into raw local cache layout, and keep legacy raw-media import compatibility - desktop-local source: supported Slack Desktop cache paths on macOS and Linux @@ -155,13 +156,14 @@ Purpose: Expected flags: -- `--source api|desktop|all` +- `--source api|bot|desktop|wiretap|mcp|connector|all|provider:` - `--workspace ` - `--channels ` - `--exclude-channels ` - `--since ` - `--full` - `--latest-only` +- `--limit ` for bounded external-provider validation imports - `--concurrency ` - `--auto-join=` @@ -310,6 +312,15 @@ Credential model: - `[sync].auto_join` defaults to `true` and controls whether API sync attempts to join public channels before retrying history - `[sync].exclude_channels` is an optional case-insensitive list of channel names to skip during API sync and merges with `--exclude-channels` +External provider config: + +- each `[[providers]]` entry has a unique lowercase `name` without whitespace, slashes, or colons +- `command` is required, expands `~` or a leading `~/`, and must resolve to an absolute path +- `args` are passed directly to the command without a shell +- `env_allowlist` names additional environment variables forwarded alongside the minimal runtime environment +- `source_rank` is required and must be greater than `2`; lower numeric ranks win during message reconciliation, while equal ranks may replace +- `batch_size` defaults to `1000`, must be between `1` and `100000`, and is forwarded as an upstream batching hint + Share config: - `[share].remote` points at the git remote that stores compressed archive snapshots @@ -352,6 +363,44 @@ Share config: 14. update FTS rows and mentions 15. write checkpoints, channel skips, and join attempts +### External provider sync + +1. resolve `provider:` against `[[providers]]` and require a workspace ID +2. choose a checkpoint key from the provider name, workspace, and normalized invocation scope + - the unfiltered incremental run uses the workspace checkpoint + - `--since`, `--full`, `--latest-only`, channel filters, exclusions, and `--limit` use isolated scope checkpoints +3. start the absolute command directly with configured args and a minimal environment plus `env_allowlist` +4. send one `slacrawl-provider-v1` JSON request on stdin containing `workspace_id`, `since`, `full`, `latest_only`, channel filters, saved opaque `checkpoint`, `batch_size`, and optional positive `limit` +5. consume JSONL from stdout + - the first record must be `hello` with the matching protocol + - data records may be `workspace`, `channel`, `user`, or `message` + - `checkpoint` records must identify the requested workspace and contain a nonempty opaque value + - the terminal `done.records` count must equal the number of data records consumed; checkpoints are not counted +6. reject cross-workspace records, missing required identities or message channels, records after `done`, and more messages than `limit` +7. build normalized message search text, extract mentions, update FTS, and preserve existing messages with a lower numeric source rank +8. when a checkpoint is present, atomically commit it with the pending record batch; committed batches and checkpoints remain resumable if the process later fails +9. require `done` plus a zero exit status for overall success +10. after a successful unbounded `--full`, promote the final full checkpoint to the matching incremental scope + +Provider v1 response records: + +- `hello`: `type`, `protocol`, optional provider implementation name +- `workspace`: `type`, requested workspace `id`, optional metadata and `raw_json` +- `channel`: `type`, `workspace_id`, `id`, optional metadata and `raw_json` +- `user`: `type`, `workspace_id`, `id`, optional profile data and `raw_json` +- `message`: `type`, `workspace_id`, `channel_id`, Slack `ts`, optional `user_id`, `thread_ts`, `text`, and `raw_json` +- `checkpoint`: `type`, `entity_type = "workspace"`, requested workspace `entity_id`, and opaque nonempty `value` +- `done`: `type`, data-record count in `records`, plus optional provider quality counters + +The provider owns the interpretation of `since`, `full`, `latest_only`, +`channels`, and `exclude_channels`; the consumer enforces workspace ownership +and the positive message limit. A message channel must already exist locally or +be emitted earlier in the stream. Unknown nonempty message user IDs are reserved +as sparse workspace-bound profiles that a later real user record can enrich. +Incremental imports enforce stored retention floors; `--full` or an explicit +`--since` older than the floor is a deliberate restore that may reintroduce +purged history. + ### Git share sync 1. clone or open the configured share repo @@ -384,6 +433,7 @@ Share config: cmd/slacrawl/ internal/cli/ internal/config/ +internal/provider/ internal/share/ internal/slackapi/ internal/slackdesktop/ diff --git a/config.example.toml b/config.example.toml index 52ef4d9..80b920d 100644 --- a/config.example.toml +++ b/config.example.toml @@ -55,6 +55,16 @@ max_pages = 250 # command = "npx" # args = ["-y", "@modelcontextprotocol/server-slack"] +# Generic local archive provider (selected with --source provider:archive). +# Keep this block commented until the executable is installed. +# [[providers]] +# name = "archive" +# command = "/usr/local/bin/archive-provider" +# args = ["provide", "--format", "jsonl"] +# env_allowlist = ["ARCHIVE_DB_PATH"] +# source_rank = 5 # must be greater than 2; lower ranks win; equal ranks may replace +# batch_size = 1000 # default 1000; valid range 1..100000 + [sync] concurrency = 4 repair_every = "30m" diff --git a/docs/configuration.md b/docs/configuration.md index fd0f632..5c83389 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -13,6 +13,7 @@ The config is designed to work with safe defaults: - Slack Desktop is enabled by default - the desktop path is auto-detected when left blank - Slack tokens are resolved from environment variables +- external providers receive only explicitly allowlisted environment additions ## Example @@ -111,8 +112,9 @@ One config/database should represent one Slack visibility boundary: the messages - `sync --source bot` is an alias for `sync --source api` and uses Slack bot/user tokens - `sync --source mcp` fetches from a Slack connector exposed by the configured HTTP JSON-RPC MCP gateway +- `sync --source provider:` imports a configured external archive through a trusted local subprocess - `sync --source wiretap` is an alias for `sync --source desktop` and reads the local Slack Desktop cache -- `sync --source all` runs token-backed sync first, then desktop enrichment +- `sync --source all` runs token-backed sync first, then desktop enrichment; external providers remain explicit - `[share]` backs up the current DB and safely merges snapshots by default; it is not a second Slack data source - exact latest or historical replacement requires `update --restore` @@ -185,6 +187,132 @@ The subprocess receives a minimal environment plus known Slack/Codex token varia Tool discovery selects either the Codex Slack connector contract or the reference `slack_list_channels`, `slack_get_channel_history`, `slack_get_thread_replies`, and `slack_get_users` contract. +## External Archive Providers + +Use `[[providers]]` to adapt another local archive to the canonical SQLite +schema without adding a source-specific integration to `slacrawl` itself. +Each provider is an explicit sync source and is not included in `--source all`. + +```toml +[[providers]] +name = "archive" +command = "/usr/local/bin/archive-provider" +args = ["provide", "--format", "jsonl"] +env_allowlist = ["ARCHIVE_DB_PATH"] +source_rank = 5 +batch_size = 1000 +``` + +```bash +slacrawl sync --source provider:archive --workspace T01234567 +slacrawl sync --source provider:archive --workspace T01234567 --channels C01234567 +slacrawl sync --source provider:archive --workspace T01234567 --full +slacrawl sync --source provider:archive --workspace T01234567 --limit 100 +``` + +Configuration rules: + +- `name` is normalized to lowercase and must be unique; whitespace, slashes, and colons are rejected +- `command` is required; `~` or a leading `~/` is expanded, then the path must be absolute +- `args` are passed directly to the executable without shell parsing +- `env_allowlist` names additional variables to copy from the parent environment; values do not belong in TOML +- the subprocess otherwise receives only ordinary runtime variables such as `HOME`, `PATH`, temporary-directory, user, shell, and platform variables when present +- `source_rank` must be greater than `2`; lower numeric ranks win, so provider messages cannot replace canonical rank `1` or `2` data; equal ranks may replace +- `batch_size` defaults to `1000`, accepts `1` through `100000`, and is forwarded to the provider as a preferred upstream batch size + +### JSONL protocol + +`slacrawl` writes exactly one JSON object followed by a newline to provider stdin, +then closes stdin. Example: + +```json +{"type":"request","protocol":"slacrawl-provider-v1","workspace_id":"T01234567","channels":["C01234567"],"checkpoint":"opaque-provider-state","batch_size":1000,"limit":100} +``` + +`limit` is omitted when no positive limit was requested. The other optional +fields may also be omitted when empty. `checkpoint` is an opaque string that +the provider previously emitted; `slacrawl` never parses it. + +Provider stdout is JSONL. Records must appear in this order: + +1. One `hello` record before any other output. +2. Zero or more data and checkpoint records. +3. Exactly one terminal `done` record, followed by EOF. + +An example response looks like: + +```jsonl +{"type":"hello","protocol":"slacrawl-provider-v1","provider":"archive-cache"} +{"type":"workspace","id":"T01234567","name":"Example Workspace","raw_json":{}} +{"type":"user","workspace_id":"T01234567","id":"U01234567","name":"example","raw_json":{}} +{"type":"channel","workspace_id":"T01234567","id":"C01234567","name":"general","kind":"public_channel","raw_json":{}} +{"type":"message","workspace_id":"T01234567","channel_id":"C01234567","ts":"1772574099.659199","user_id":"U01234567","thread_ts":"","text":"Example message","raw_json":{}} +{"type":"checkpoint","entity_type":"workspace","entity_id":"T01234567","value":"opaque-next-state"} +{"type":"done","records":4} +``` + +`done.records` counts `workspace`, `channel`, `user`, and `message` records; it +does not count `hello`, `checkpoint`, or `done`. Optional fields by record type: + +- `workspace`: `name`, `domain`, `enterprise_id`, `raw_json` +- `channel`: `name`, `kind`, `topic`, `purpose`, `is_private`, `is_archived`, `is_shared`, `is_general`, `raw_json` +- `user`: `name`, `real_name`, `display_name`, `title`, `is_bot`, `is_deleted`, `raw_json` +- `message`: `user_id`, `thread_ts`, `text`, `raw_json` +- `done`: `skipped_users`, `skipped_channels`, `skipped_bad_message_id`, `skipped_duplicate_identity`, `skipped_missing_channel`, `skipped_missing_timestamp_sort_key`, `rounded_thread_messages`, `rounded_threads`, `unresolved_thread_messages`, and `unresolved_threads` + +Unknown JSON fields are ignored. + +Required identity fields are the requested workspace `id`; channel and user +`workspace_id` plus `id`; and message `workspace_id`, `channel_id`, and Slack +`ts`. A message channel must already exist in the database or have appeared +earlier in the stream. An unknown nonempty message user ID is preserved as a +sparse workspace-bound profile that a later real user record may enrich. Empty +names fall back to their IDs, empty channel kinds become `provider_channel`, +and missing or `null` `raw_json` becomes `{}`. + +`slacrawl` builds repaired, normalized search text from each message, extracts +mentions, updates FTS, and uses `source_rank` reconciliation for canonical +message rows. Metadata emitted by a provider is sparse enrichment and does not +replace richer existing workspace, channel, or user metadata. + +### Checkpoints, scopes, and limits + +The normal unfiltered incremental run stores one checkpoint under +`provider:` and the workspace ID. Any `--since`, `--full`, +`--latest-only`, channel filter, exclusion, or `--limit` gets a deterministic +scope-specific checkpoint. This prevents a bounded validation import or a +partial backfill from advancing the normal incremental cursor. + +A `checkpoint` record flushes pending records and the checkpoint in one SQLite +transaction. Emit it only after all records covered by its value. If the +provider later exits nonzero or omits `done`, the command fails, but already +committed records and checkpoints remain available for the next retry. A +successful unbounded `--full` promotes its last checkpoint to the matching +incremental scope; a limited full run does not. + +`--limit` is intended for validation imports. `slacrawl` forwards a positive +message limit and fails if the provider emits more messages. The limit applies +only to `message` records, not catalog records. Providers must also interpret +and honor `since`, `full`, `latest_only`, `channels`, and `exclude_channels`; +`slacrawl` forwards those values but cannot infer source-specific filtering. + +Incremental provider imports enforce stored channel retention floors. `--full`, +or an explicit `--since` older than a channel's floor, is treated as a deliberate +restore and may reintroduce purged history. + +### Safety and failure behavior + +- treat the executable as trusted code inside the archive visibility boundary +- `slacrawl` restricts environment forwarding but does not sandbox the executable; it retains the caller's filesystem and network permissions +- allowlist only variables the provider needs, especially credentials or archive paths +- keep separate configs and databases for archives with different visibility +- the first record must be `hello` with protocol `slacrawl-provider-v1` +- cross-workspace records, missing required identities or channels, protocol mismatches, output after `done`, and record-count mismatches fail closed +- the process must emit `done`, close stdout, and exit zero for the run to succeed +- EOF without `done` flushes pending records before reporting failure; exceeding `--limit` likewise flushes pending records through the limit before failing +- provider stderr is capped and attached to failures for diagnostics +- provider v1 transfers metadata and message text, not file blobs + ## Git Archive Sharing Use `[share]` when you want one machine to publish a private Slack archive snapshot and other machines to query it locally without Slack API credentials. diff --git a/internal/cli/app.go b/internal/cli/app.go index 58c369f..ed43bff 100644 --- a/internal/cli/app.go +++ b/internal/cli/app.go @@ -521,19 +521,23 @@ func (a *App) runSync(ctx context.Context, configPath string, args []string, for fs := flag.NewFlagSet("sync", flag.ContinueOnError) fs.SetOutput(a.Stderr) - source := fs.String("source", "api", "api|bot|desktop|wiretap|mcp|connector|all") + source := fs.String("source", "api", "api|bot|desktop|wiretap|mcp|connector|all|provider:") workspaceID := fs.String("workspace", "", "workspace id") channels := fs.String("channels", "", "comma separated channel ids") excludeChannels := fs.String("exclude-channels", "", "comma separated channel names to skip during sync") since := fs.String("since", "", "oldest slack ts or RFC3339 timestamp") full := fs.Bool("full", false, "full sync") latestOnly := fs.Bool("latest-only", false, "skip first-time historical backfills") + limit := fs.Int("limit", 0, "maximum provider messages (validation imports only)") concurrency := fs.Int("concurrency", cfg.Sync.Concurrency, "worker count") withMedia := fs.Bool("with-media", cfg.FileMediaEnabled(), "fetch file media after sync") autoJoin := fs.Bool("auto-join", cfg.Sync.AutoJoinResolved(), "auto-join public channels during sync") if err := fs.Parse(args); err != nil { return err } + if *limit < 0 { + return errors.New("limit cannot be negative") + } resolvedSource, err := syncer.ParseSource(*source) if err != nil { @@ -553,6 +557,7 @@ func (a *App) runSync(ctx context.Context, configPath string, args []string, for Since: *since, Full: *full, LatestOnly: *latestOnly, + Limit: *limit, Concurrency: *concurrency, AutoJoin: boolPtr(*autoJoin), APIURL: a.apiURL, diff --git a/internal/config/config.go b/internal/config/config.go index 3d7fbb3..9d7a6d4 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -24,6 +24,7 @@ type Config struct { Version int `toml:"version"` WorkspaceID string `toml:"workspace_id"` Workspaces []Workspace `toml:"workspaces"` + Providers []Provider `toml:"providers"` DBPath string `toml:"db_path"` CacheDir string `toml:"cache_dir"` LogDir string `toml:"log_dir"` @@ -33,6 +34,15 @@ type Config struct { Share ShareConfig `toml:"share"` } +type Provider struct { + Name string `toml:"name"` + Command string `toml:"command"` + Args []string `toml:"args"` + EnvAllowlist []string `toml:"env_allowlist"` + SourceRank int `toml:"source_rank"` + BatchSize int `toml:"batch_size"` +} + type SlackConfig struct { Bot TokenConfig `toml:"bot"` App TokenConfig `toml:"app"` @@ -292,6 +302,39 @@ func (c *Config) Normalize() error { } c.Workspaces[i].ID = strings.TrimSpace(c.Workspaces[i].ID) } + providerNames := make(map[string]struct{}, len(c.Providers)) + for i := range c.Providers { + provider := &c.Providers[i] + provider.Name = strings.ToLower(strings.TrimSpace(provider.Name)) + if provider.Name == "" { + return fmt.Errorf("providers[%d].name is required", i) + } + if strings.ContainsAny(provider.Name, ":/\\\t\r\n ") { + return fmt.Errorf("providers[%d].name %q must not contain whitespace, slashes, or colons", i, provider.Name) + } + if _, exists := providerNames[provider.Name]; exists { + return fmt.Errorf("duplicate provider name %q", provider.Name) + } + providerNames[provider.Name] = struct{}{} + provider.Command = crawlconfig.ExpandHome(strings.TrimSpace(provider.Command)) + if provider.Command == "" { + return fmt.Errorf("providers[%d].command is required", i) + } + if !filepath.IsAbs(provider.Command) { + return fmt.Errorf("providers[%d].command must be an absolute path", i) + } + provider.Command = filepath.Clean(provider.Command) + provider.EnvAllowlist = normalizeStringList(provider.EnvAllowlist) + if provider.SourceRank <= 2 { + return fmt.Errorf("providers[%d].source_rank must be greater than 2", i) + } + if provider.BatchSize == 0 { + provider.BatchSize = 1_000 + } + if provider.BatchSize < 1 || provider.BatchSize > 100_000 { + return fmt.Errorf("providers[%d].batch_size must be between 1 and 100000", i) + } + } c.WorkspaceID = strings.TrimSpace(c.WorkspaceID) if c.WorkspaceID == "" { c.WorkspaceID = c.DefaultWorkspaceID() @@ -299,6 +342,16 @@ func (c *Config) Normalize() error { return nil } +func (c Config) Provider(name string) (Provider, bool) { + name = strings.ToLower(strings.TrimSpace(name)) + for _, provider := range c.Providers { + if provider.Name == name { + return provider, true + } + } + return Provider{}, false +} + func normalizeStringList(values []string) []string { out := make([]string, 0, len(values)) seen := map[string]bool{} diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 342e1f9..0e421f8 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -103,6 +103,10 @@ func TestSaveAndLoadRoundTrip(t *testing.T) { cfg.Slack.MCP.Enabled = true cfg.Slack.MCP.BaseURL = "https://mcp.example.test" cfg.Slack.MCP.AuthPath = filepath.Join(dir, "auth.json") + cfg.Providers = []Provider{{ + Name: " Archive ", Command: filepath.Join(dir, "archive-provider"), Args: []string{"provide"}, + EnvAllowlist: []string{"ARCHIVE_CACHE_PATH", "ARCHIVE_CACHE_PATH"}, SourceRank: 5, + }} require.NoError(t, cfg.Save(path)) loaded, err := Load(path) @@ -116,6 +120,34 @@ func TestSaveAndLoadRoundTrip(t *testing.T) { require.True(t, filepath.IsAbs(loaded.DBPath)) require.True(t, filepath.IsAbs(loaded.Share.RepoPath)) require.Equal(t, "https://example.com/private/slacrawl.git", loaded.Share.Remote) + require.Len(t, loaded.Providers, 1) + require.Equal(t, "archive", loaded.Providers[0].Name) + require.Equal(t, 1_000, loaded.Providers[0].BatchSize) + require.Equal(t, []string{"ARCHIVE_CACHE_PATH"}, loaded.Providers[0].EnvAllowlist) + provider, ok := loaded.Provider("ARCHIVE") + require.True(t, ok) + require.Equal(t, filepath.Join(dir, "archive-provider"), provider.Command) +} + +func TestNormalizeRejectsUnsafeProviderConfig(t *testing.T) { + tests := []struct { + name string + provider Provider + message string + }{ + {name: "relative command", provider: Provider{Name: "p", Command: "archive-provider", SourceRank: 5}, message: "absolute path"}, + {name: "priority one", provider: Provider{Name: "p", Command: "/bin/true", SourceRank: 1}, message: "greater than 2"}, + {name: "priority tie", provider: Provider{Name: "p", Command: "/bin/true", SourceRank: 2}, message: "greater than 2"}, + {name: "name", provider: Provider{Name: "bad:name", Command: "/bin/true", SourceRank: 5}, message: "must not contain"}, + {name: "batch", provider: Provider{Name: "p", Command: "/bin/true", SourceRank: 5, BatchSize: 100_001}, message: "between 1 and 100000"}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + cfg := Default() + cfg.Providers = []Provider{test.provider} + require.ErrorContains(t, cfg.Normalize(), test.message) + }) + } } func TestExpandPathMakesRelativePathsAbsolute(t *testing.T) { diff --git a/internal/provider/provider.go b/internal/provider/provider.go new file mode 100644 index 0000000..f222027 --- /dev/null +++ b/internal/provider/provider.go @@ -0,0 +1,666 @@ +package provider + +import ( + "bytes" + "context" + "crypto/sha256" + "database/sql" + "encoding/json" + "errors" + "fmt" + "io" + "log/slog" + "os" + "os/exec" + "sort" + "strings" + "time" + + "github.com/openclaw/slacrawl/internal/config" + "github.com/openclaw/slacrawl/internal/search" + "github.com/openclaw/slacrawl/internal/store" + "github.com/slack-go/slack" +) + +const ProtocolVersion = "slacrawl-provider-v1" + +const maxStoreBatchRecords = 1_000 + +type Options struct { + WorkspaceID string + Channels []string + ExcludeChannels []string + Since string + Full bool + LatestOnly bool + Limit int + Logger *slog.Logger +} + +type Summary struct { + Name string `json:"name"` + Provider string `json:"provider,omitempty"` + WorkspaceID string `json:"workspace_id"` + Records int64 `json:"records"` + Workspaces int64 `json:"workspaces"` + Channels int64 `json:"channels"` + Users int64 `json:"users"` + Messages int64 `json:"messages"` + MessagesWritten int64 `json:"messages_written"` + Checkpoints int64 `json:"checkpoints"` + SkippedUsers int64 `json:"skipped_users,omitempty"` + SkippedChannels int64 `json:"skipped_channels,omitempty"` + SkippedBadMessageID int64 `json:"skipped_bad_message_id,omitempty"` + SkippedDuplicateIdentity int64 `json:"skipped_duplicate_identity,omitempty"` + SkippedMissingChannel int64 `json:"skipped_missing_channel,omitempty"` + SkippedMissingSortKey int64 `json:"skipped_missing_timestamp_sort_key,omitempty"` + RoundedThreadMessages int64 `json:"rounded_thread_messages,omitempty"` + RoundedThreads int64 `json:"rounded_threads,omitempty"` + UnresolvedThreadMessages int64 `json:"unresolved_thread_messages,omitempty"` + UnresolvedThreads int64 `json:"unresolved_threads,omitempty"` +} + +type request struct { + Type string `json:"type"` + Protocol string `json:"protocol"` + WorkspaceID string `json:"workspace_id"` + Since string `json:"since,omitempty"` + Full bool `json:"full,omitempty"` + LatestOnly bool `json:"latest_only,omitempty"` + Channels []string `json:"channels,omitempty"` + ExcludeChannels []string `json:"exclude_channels,omitempty"` + Checkpoint string `json:"checkpoint,omitempty"` + BatchSize int `json:"batch_size"` + Limit *int `json:"limit,omitempty"` +} + +type record struct { + Type string `json:"type"` + Protocol string `json:"protocol"` + Provider string `json:"provider"` + WorkspaceID string `json:"workspace_id"` + ID string `json:"id"` + Name string `json:"name"` + Domain string `json:"domain"` + EnterpriseID string `json:"enterprise_id"` + RealName string `json:"real_name"` + DisplayName string `json:"display_name"` + Title string `json:"title"` + Kind string `json:"kind"` + Topic string `json:"topic"` + Purpose string `json:"purpose"` + IsPrivate bool `json:"is_private"` + IsArchived bool `json:"is_archived"` + IsShared bool `json:"is_shared"` + IsGeneral bool `json:"is_general"` + IsBot bool `json:"is_bot"` + IsDeleted bool `json:"is_deleted"` + ChannelID string `json:"channel_id"` + TS string `json:"ts"` + UserID string `json:"user_id"` + ThreadTS string `json:"thread_ts"` + Text string `json:"text"` + RawJSON json.RawMessage `json:"raw_json"` + EntityType string `json:"entity_type"` + EntityID string `json:"entity_id"` + Value string `json:"value"` + Records int64 `json:"records"` + + SkippedUsers int64 `json:"skipped_users"` + SkippedChannels int64 `json:"skipped_channels"` + SkippedBadMessageID int64 `json:"skipped_bad_message_id"` + SkippedDuplicateIdentity int64 `json:"skipped_duplicate_identity"` + SkippedMissingChannel int64 `json:"skipped_missing_channel"` + SkippedMissingSortKey int64 `json:"skipped_missing_timestamp_sort_key"` + RoundedThreadMessages int64 `json:"rounded_thread_messages"` + RoundedThreads int64 `json:"rounded_threads"` + UnresolvedThreadMessages int64 `json:"unresolved_thread_messages"` + UnresolvedThreads int64 `json:"unresolved_threads"` +} + +func SourceName(name string) string { + return "provider:" + strings.ToLower(strings.TrimSpace(name)) +} + +func Sync(ctx context.Context, st *store.Store, providerConfig config.Provider, opts Options) (Summary, error) { + workspaceID := strings.TrimSpace(opts.WorkspaceID) + if workspaceID == "" { + return Summary{}, errors.New("workspace ID is required for provider sync") + } + if opts.Limit < 0 { + return Summary{}, errors.New("provider limit cannot be negative") + } + if providerConfig.SourceRank <= 2 { + return Summary{}, errors.New("provider source rank must be greater than 2") + } + sourceName := SourceName(providerConfig.Name) + checkpointKey := stateKey(workspaceID, opts) + checkpoint, err := st.GetSyncState(ctx, sourceName, checkpointKey.entityType, checkpointKey.entityID) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return Summary{}, fmt.Errorf("read provider checkpoint: %w", err) + } + if errors.Is(err, sql.ErrNoRows) { + checkpoint = "" + } + providerRequest := request{ + Type: "request", + Protocol: ProtocolVersion, + WorkspaceID: workspaceID, + Since: strings.TrimSpace(opts.Since), + Full: opts.Full, + LatestOnly: opts.LatestOnly, + Channels: opts.Channels, + ExcludeChannels: opts.ExcludeChannels, + Checkpoint: checkpoint, + BatchSize: providerConfig.BatchSize, + } + if opts.Limit > 0 { + providerRequest.Limit = &opts.Limit + } + + runCtx, cancel := context.WithCancel(ctx) + defer cancel() + cmd := exec.CommandContext(runCtx, providerConfig.Command, providerConfig.Args...) + cmd.Env = providerEnvironment(providerConfig.EnvAllowlist) + stdin, err := cmd.StdinPipe() + if err != nil { + return Summary{}, fmt.Errorf("open provider stdin: %w", err) + } + stdout, err := cmd.StdoutPipe() + if err != nil { + _ = stdin.Close() + return Summary{}, fmt.Errorf("open provider stdout: %w", err) + } + var stderr cappedBuffer + stderr.limit = 64 << 10 + cmd.Stderr = &stderr + if err := cmd.Start(); err != nil { + _ = stdin.Close() + return Summary{}, fmt.Errorf("start provider %q: %w", providerConfig.Name, err) + } + if err := json.NewEncoder(stdin).Encode(providerRequest); err != nil { + _ = stdin.Close() + cancel() + _ = cmd.Wait() + return Summary{}, fmt.Errorf("write provider request: %w", err) + } + if err := stdin.Close(); err != nil { + cancel() + _ = cmd.Wait() + return Summary{}, fmt.Errorf("close provider request: %w", err) + } + + consumer := consumer{ + store: st, + config: providerConfig, + opts: opts, + sourceName: sourceName, + workspaceID: workspaceID, + checkpointKey: checkpointKey, + now: time.Now().UTC(), + validChannels: make(map[string]struct{}), + validUsers: make(map[string]struct{}), + retention: make(map[string]bool), + summary: Summary{ + Name: providerConfig.Name, + WorkspaceID: workspaceID, + }, + } + consumeErr := consumer.consume(ctx, stdout) + if consumeErr != nil { + cancel() + } + waitErr := cmd.Wait() + if consumeErr != nil { + return consumer.summary, withStderr(fmt.Errorf("consume provider %q: %w", providerConfig.Name, consumeErr), stderr.String()) + } + if waitErr != nil { + return consumer.summary, withStderr(fmt.Errorf("provider %q exited: %w", providerConfig.Name, waitErr), stderr.String()) + } + if !consumer.done { + return consumer.summary, withStderr(fmt.Errorf("provider %q ended without done record", providerConfig.Name), stderr.String()) + } + if err := consumer.promoteCompletedFullCheckpoint(ctx); err != nil { + return consumer.summary, err + } + return consumer.summary, nil +} + +type consumer struct { + store *store.Store + config config.Provider + opts Options + sourceName string + workspaceID string + checkpointKey checkpointStateKey + now time.Time + summary Summary + hello bool + done bool + counted int64 + lastCheckpoint string + validChannels map[string]struct{} + validUsers map[string]struct{} + retention map[string]bool + batch store.WriteBatch + seenMessages int64 +} + +func (c *consumer) consume(ctx context.Context, input io.Reader) error { + decoder := json.NewDecoder(input) + for { + var item record + if err := decoder.Decode(&item); errors.Is(err, io.EOF) { + return c.flush(ctx) + } else if err != nil { + return fmt.Errorf("decode provider record: %w", err) + } + if c.done { + return errors.New("provider emitted a record after done") + } + if !c.hello && item.Type != "hello" { + return fmt.Errorf("provider first record must be hello, got %q", item.Type) + } + switch item.Type { + case "hello": + if c.hello { + return errors.New("provider emitted duplicate hello") + } + if item.Protocol != ProtocolVersion { + return fmt.Errorf("provider protocol mismatch: expected %q, got %q", ProtocolVersion, item.Protocol) + } + c.hello = true + c.summary.Provider = item.Provider + case "workspace": + if item.ID != c.workspaceID { + return fmt.Errorf("provider workspace record %q does not match requested workspace %q", item.ID, c.workspaceID) + } + name := strings.TrimSpace(item.Name) + if name == "" { + name = item.ID + } + c.batch.Workspaces = append(c.batch.Workspaces, store.Workspace{ + ID: item.ID, Name: name, Domain: item.Domain, EnterpriseID: item.EnterpriseID, + RawJSON: rawJSON(item.RawJSON), UpdatedAt: c.now, + }) + c.counted++ + if err := c.flushIfFull(ctx); err != nil { + return err + } + case "channel": + if err := c.validateWorkspace(item.WorkspaceID, "channel", item.ID); err != nil { + return err + } + name := strings.TrimSpace(item.Name) + if name == "" { + name = item.ID + } + kind := strings.TrimSpace(item.Kind) + if kind == "" { + kind = "provider_channel" + } + c.batch.Channels = append(c.batch.Channels, store.Channel{ + ID: item.ID, WorkspaceID: item.WorkspaceID, Name: name, Kind: kind, + Topic: item.Topic, Purpose: item.Purpose, IsPrivate: item.IsPrivate, + IsArchived: item.IsArchived, IsShared: item.IsShared, IsGeneral: item.IsGeneral, + RawJSON: rawJSON(item.RawJSON), UpdatedAt: c.now, + }) + c.validChannels[item.ID] = struct{}{} + c.counted++ + if err := c.flushIfFull(ctx); err != nil { + return err + } + case "user": + if err := c.validateWorkspace(item.WorkspaceID, "user", item.ID); err != nil { + return err + } + name := strings.TrimSpace(item.Name) + if name == "" { + name = item.ID + } + c.batch.Users = append(c.batch.Users, store.User{ + ID: item.ID, WorkspaceID: item.WorkspaceID, Name: name, RealName: item.RealName, + DisplayName: item.DisplayName, Title: item.Title, IsBot: item.IsBot, + IsDeleted: item.IsDeleted, RawJSON: rawJSON(item.RawJSON), UpdatedAt: c.now, + }) + c.validUsers[item.ID] = struct{}{} + c.counted++ + if err := c.flushIfFull(ctx); err != nil { + return err + } + case "message": + if c.opts.Limit > 0 && c.seenMessages >= int64(c.opts.Limit) { + if err := c.flush(ctx); err != nil { + return err + } + return fmt.Errorf("provider emitted more than requested message limit %d", c.opts.Limit) + } + if err := c.applyMessage(ctx, item); err != nil { + return err + } + c.seenMessages++ + c.counted++ + if err := c.flushIfFull(ctx); err != nil { + return err + } + if c.opts.Logger != nil && c.seenMessages%100_000 == 0 { + c.opts.Logger.Info("provider sync progress", "provider", c.config.Name, "messages", c.seenMessages) + } + case "checkpoint": + if item.EntityType != "workspace" || item.EntityID != c.workspaceID { + return fmt.Errorf("provider checkpoint must target workspace %q", c.workspaceID) + } + if strings.TrimSpace(item.Value) == "" { + return errors.New("provider checkpoint value is required") + } + c.batch.SyncStates = append(c.batch.SyncStates, store.SyncStateWrite{ + SourceName: c.sourceName, EntityType: c.checkpointKey.entityType, + EntityID: c.checkpointKey.entityID, Value: item.Value, + }) + if err := c.flush(ctx); err != nil { + return fmt.Errorf("apply provider checkpoint: %w", err) + } + case "done": + if item.Records != c.counted { + return fmt.Errorf("provider record count mismatch: done reported %d, consumed %d", item.Records, c.counted) + } + if err := c.flush(ctx); err != nil { + return fmt.Errorf("flush provider records: %w", err) + } + c.summary.Records = item.Records + c.summary.SkippedUsers = item.SkippedUsers + c.summary.SkippedChannels = item.SkippedChannels + c.summary.SkippedBadMessageID = item.SkippedBadMessageID + c.summary.SkippedDuplicateIdentity = item.SkippedDuplicateIdentity + c.summary.SkippedMissingChannel = item.SkippedMissingChannel + c.summary.SkippedMissingSortKey = item.SkippedMissingSortKey + c.summary.RoundedThreadMessages = item.RoundedThreadMessages + c.summary.RoundedThreads = item.RoundedThreads + c.summary.UnresolvedThreadMessages = item.UnresolvedThreadMessages + c.summary.UnresolvedThreads = item.UnresolvedThreads + c.done = true + default: + return fmt.Errorf("unsupported provider record type %q", item.Type) + } + } +} + +type checkpointStateKey struct { + entityType string + entityID string +} + +func stateKey(workspaceID string, opts Options) checkpointStateKey { + scope := struct { + Since string `json:"since"` + Full bool `json:"full"` + LatestOnly bool `json:"latest_only"` + Channels []string `json:"channels"` + ExcludeChannels []string `json:"exclude_channels"` + Limit int `json:"limit"` + }{ + Since: strings.TrimSpace(opts.Since), Full: opts.Full, LatestOnly: opts.LatestOnly, + Channels: normalizedScopeValues(opts.Channels), + ExcludeChannels: normalizedScopeValues(opts.ExcludeChannels), + Limit: opts.Limit, + } + if scope.Since == "" && !scope.Full && !scope.LatestOnly && len(scope.Channels) == 0 && len(scope.ExcludeChannels) == 0 && scope.Limit == 0 { + return checkpointStateKey{entityType: "workspace", entityID: workspaceID} + } + raw, _ := json.Marshal(scope) + digest := sha256.Sum256(raw) + return checkpointStateKey{ + entityType: "workspace_scope", + entityID: fmt.Sprintf("%s|%x", workspaceID, digest[:16]), + } +} + +func normalizedScopeValues(values []string) []string { + seen := make(map[string]struct{}, len(values)) + out := make([]string, 0, len(values)) + for _, value := range values { + value = strings.TrimSpace(value) + if value == "" { + continue + } + if _, ok := seen[value]; ok { + continue + } + seen[value] = struct{}{} + out = append(out, value) + } + sort.Strings(out) + return out +} + +func (c *consumer) promoteCompletedFullCheckpoint(ctx context.Context) error { + if !c.opts.Full || c.opts.Limit > 0 || c.lastCheckpoint == "" { + return nil + } + incremental := c.opts + incremental.Full = false + key := stateKey(c.workspaceID, incremental) + if err := c.store.SetSyncState(ctx, c.sourceName, key.entityType, key.entityID, c.lastCheckpoint); err != nil { + return fmt.Errorf("promote completed full provider checkpoint: %w", err) + } + return nil +} + +func (c *consumer) validateWorkspace(workspaceID, entity, id string) error { + if strings.TrimSpace(id) == "" { + return fmt.Errorf("provider %s id is required", entity) + } + if workspaceID != c.workspaceID { + return fmt.Errorf("provider %s %q belongs to workspace %q, not requested workspace %q", entity, id, workspaceID, c.workspaceID) + } + return nil +} + +func (c *consumer) applyMessage(ctx context.Context, item record) error { + if strings.TrimSpace(item.ChannelID) == "" || strings.TrimSpace(item.TS) == "" { + return errors.New("provider message channel_id and ts are required") + } + if item.WorkspaceID != c.workspaceID { + return fmt.Errorf("provider message %q belongs to workspace %q, not requested workspace %q", item.ChannelID+"|"+item.TS, item.WorkspaceID, c.workspaceID) + } + if err := c.validateMessageReferences(ctx, item.ChannelID, item.UserID); err != nil { + return err + } + threadTS := strings.TrimSpace(item.ThreadTS) + if threadTS == item.TS { + threadTS = "" + } + slackMessage := slack.Message{Msg: slack.Msg{ + Channel: item.ChannelID, Timestamp: item.TS, ThreadTimestamp: threadTS, + User: item.UserID, Text: item.Text, + }} + mentions := search.ExtractMentions(item.Text) + storedMentions := make([]store.Mention, 0, len(mentions)) + for _, mention := range mentions { + storedMentions = append(storedMentions, store.Mention{ + Type: mention.Type, TargetID: mention.TargetID, DisplayText: mention.DisplayText, + }) + } + message := store.Message{ + ChannelID: item.ChannelID, TS: item.TS, WorkspaceID: item.WorkspaceID, + UserID: item.UserID, ThreadTS: threadTS, Text: item.Text, + NormalizedText: search.NormalizeMessage(slackMessage), SourceRank: c.config.SourceRank, + SourceName: c.sourceName, RawJSON: rawJSON(item.RawJSON), UpdatedAt: c.now, + Files: nil, + } + enforceRetention, err := c.enforceRetention(ctx, item.ChannelID) + if err != nil { + return fmt.Errorf("read provider message retention for %s: %w", item.ChannelID, err) + } + c.batch.Messages = append(c.batch.Messages, store.MessageWrite{ + Message: message, Mentions: storedMentions, + PreserveHigherPriority: true, EnforceRetention: enforceRetention, SkipUnchangedProviderRow: true, + }) + return nil +} + +func (c *consumer) validateMessageReferences(ctx context.Context, channelID, userID string) error { + if _, ok := c.validChannels[channelID]; !ok { + workspaceID, err := c.store.ChannelWorkspaceID(ctx, channelID) + if errors.Is(err, sql.ErrNoRows) { + return fmt.Errorf("provider message references missing channel %q", channelID) + } + if err != nil { + return fmt.Errorf("read provider message channel %q: %w", channelID, err) + } + if workspaceID != c.workspaceID { + return fmt.Errorf("provider message channel %q belongs to workspace %q, not requested workspace %q", channelID, workspaceID, c.workspaceID) + } + c.validChannels[channelID] = struct{}{} + } + userID = strings.TrimSpace(userID) + if userID == "" { + return nil + } + if _, ok := c.validUsers[userID]; ok { + return nil + } + workspaceID, err := c.store.UserWorkspaceID(ctx, userID) + if errors.Is(err, sql.ErrNoRows) { + // Reserve the global Slack ID in this workspace. A later real catalog record + // replaces this marker without letting another workspace claim the author. + c.batch.Users = append(c.batch.Users, store.User{ + ID: userID, WorkspaceID: c.workspaceID, Name: userID, + RawJSON: store.PlaceholderUserRawJSON, UpdatedAt: c.now, + }) + c.validUsers[userID] = struct{}{} + return nil + } + if err != nil { + return fmt.Errorf("read provider message user %q: %w", userID, err) + } + if workspaceID != c.workspaceID { + return fmt.Errorf("provider message user %q belongs to workspace %q, not requested workspace %q", userID, workspaceID, c.workspaceID) + } + c.validUsers[userID] = struct{}{} + return nil +} + +func (c *consumer) enforceRetention(ctx context.Context, channelID string) (bool, error) { + if c.opts.Full { + return false, nil + } + if strings.TrimSpace(c.opts.Since) == "" { + return true, nil + } + if enforce, ok := c.retention[channelID]; ok { + return enforce, nil + } + floor, err := c.store.ChannelRetentionFloor(ctx, c.workspaceID, channelID) + if err != nil { + return false, err + } + enforce := store.ShouldEnforceRetention(c.opts.Since, floor, true) + c.retention[channelID] = enforce + return enforce, nil +} + +func (c *consumer) pendingRecords() int { + return len(c.batch.Workspaces) + len(c.batch.Channels) + len(c.batch.Users) + len(c.batch.Messages) +} + +func (c *consumer) flushIfFull(ctx context.Context) error { + if c.pendingRecords() < maxStoreBatchRecords { + return nil + } + return c.flush(ctx) +} + +func (c *consumer) flush(ctx context.Context) error { + if c.pendingRecords() == 0 && len(c.batch.SyncStates) == 0 { + return nil + } + workspaces := len(c.batch.Workspaces) + channels := len(c.batch.Channels) + users := len(c.batch.Users) + messages := len(c.batch.Messages) + checkpoints := len(c.batch.SyncStates) + lastCheckpoint := "" + if checkpoints > 0 { + lastCheckpoint = c.batch.SyncStates[checkpoints-1].Value + } + result, err := c.store.ApplyWriteBatch(ctx, c.batch) + if err != nil { + return err + } + c.summary.Workspaces += int64(workspaces) + c.summary.Channels += int64(channels) + c.summary.Users += int64(users) + c.summary.Messages += int64(messages) + c.summary.MessagesWritten += int64(result.MessagesWritten) + c.summary.Checkpoints += int64(checkpoints) + if lastCheckpoint != "" { + c.lastCheckpoint = lastCheckpoint + } + c.batch = store.WriteBatch{} + return nil +} + +func rawJSON(value json.RawMessage) string { + trimmed := bytes.TrimSpace(value) + if len(trimmed) == 0 || bytes.Equal(trimmed, []byte("null")) { + return "{}" + } + return string(trimmed) +} + +func providerEnvironment(allowlist []string) []string { + keys := append([]string{ + "COMSPEC", "HOME", "LOGNAME", "PATH", "PATHEXT", "SHELL", "SYSTEMROOT", + "TEMP", "TMP", "TMPDIR", "USER", "USERNAME", "WINDIR", + }, allowlist...) + seen := make(map[string]struct{}, len(keys)) + env := make([]string, 0, len(keys)) + for _, key := range keys { + key = strings.TrimSpace(key) + if key == "" { + continue + } + if _, ok := seen[key]; ok { + continue + } + seen[key] = struct{}{} + if value, ok := os.LookupEnv(key); ok { + env = append(env, key+"="+value) + } + } + return env +} + +type cappedBuffer struct { + bytes.Buffer + limit int +} + +func (b *cappedBuffer) Write(value []byte) (int, error) { + original := len(value) + if b.limit <= 0 { + return original, nil + } + if len(value) >= b.limit { + b.Buffer.Reset() + _, _ = b.Buffer.Write(value[len(value)-b.limit:]) + return original, nil + } + if b.Buffer.Len()+len(value) > b.limit { + current := b.Buffer.Bytes() + keep := b.limit - len(value) + copy(current, current[len(current)-keep:]) + b.Buffer.Truncate(keep) + } + _, _ = b.Buffer.Write(value) + return original, nil +} + +func withStderr(err error, stderr string) error { + stderr = strings.TrimSpace(stderr) + if stderr == "" { + return err + } + return fmt.Errorf("%w: %s", err, stderr) +} diff --git a/internal/provider/provider_test.go b/internal/provider/provider_test.go new file mode 100644 index 0000000..8621b07 --- /dev/null +++ b/internal/provider/provider_test.go @@ -0,0 +1,334 @@ +package provider + +import ( + "context" + "encoding/json" + "fmt" + "os" + "path/filepath" + "testing" + "time" + + "github.com/openclaw/slacrawl/internal/config" + "github.com/openclaw/slacrawl/internal/store" + "github.com/stretchr/testify/require" +) + +func TestSyncAppliesRecordsPreservesPriorityAndPersistsCheckpoint(t *testing.T) { + st := openTestStore(t) + ctx := context.Background() + now := time.Now().UTC() + require.NoError(t, st.UpsertWorkspace(ctx, store.Workspace{ID: "T1", Name: "rich workspace", RawJSON: `{"rich":true}`, UpdatedAt: now})) + require.NoError(t, st.UpsertChannel(ctx, store.Channel{ID: "C1", WorkspaceID: "T1", Name: "rich-channel", Kind: "private_channel", RawJSON: `{"rich":true}`, UpdatedAt: now})) + require.NoError(t, st.UpsertUser(ctx, store.User{ID: "U1", WorkspaceID: "T1", Name: "rich-user", RawJSON: `{"rich":true}`, UpdatedAt: now})) + require.NoError(t, st.UpsertMessage(ctx, store.Message{ + ChannelID: "C1", TS: "2.000002", WorkspaceID: "T1", UserID: "U1", Text: "native", + NormalizedText: "native", SourceRank: 2, SourceName: "api-bot", RawJSON: `{"native":true}`, UpdatedAt: now, + }, nil)) + require.NoError(t, st.SetSyncState(ctx, "provider:fixture", "workspace", "T1", "old-checkpoint")) + + requestPath := filepath.Join(t.TempDir(), "request.json") + summary, err := Sync(ctx, st, helperProvider(t, "success", requestPath), Options{ + WorkspaceID: "T1", Channels: []string{"C1"}, ExcludeChannels: []string{"skip"}, + Since: "1.000001", Full: true, Limit: 2, + }) + require.NoError(t, err) + require.Equal(t, int64(5), summary.Records) + require.Equal(t, int64(2), summary.Messages) + require.Equal(t, int64(1), summary.MessagesWritten) + + var sent request + require.NoError(t, json.Unmarshal(requireReadFile(t, requestPath), &sent)) + require.Empty(t, sent.Checkpoint) + require.True(t, sent.Full) + require.Equal(t, "1.000001", sent.Since) + require.Equal(t, []string{"C1"}, sent.Channels) + require.Equal(t, []string{"skip"}, sent.ExcludeChannels) + require.NotNil(t, sent.Limit) + require.Equal(t, 2, *sent.Limit) + require.Equal(t, 7, sent.BatchSize) + + checkpoint, err := st.GetSyncState(ctx, "provider:fixture", "workspace", "T1") + require.NoError(t, err) + require.Equal(t, "old-checkpoint", checkpoint) + boundedKey := stateKey("T1", Options{ + WorkspaceID: "T1", Channels: []string{"C1"}, ExcludeChannels: []string{"skip"}, + Since: "1.000001", Full: true, Limit: 2, + }) + boundedCheckpoint, err := st.GetSyncState(ctx, "provider:fixture", boundedKey.entityType, boundedKey.entityID) + require.NoError(t, err) + require.Equal(t, `{"cursor":"done"}`, boundedCheckpoint) + rows, err := st.QueryReadOnly(ctx, ` +select ts, text, normalized_text, source_name, source_rank +from messages where channel_id = 'C1' order by ts`) + require.NoError(t, err) + require.Len(t, rows, 2) + require.Equal(t, "provider text <@U1>", rows[0]["text"]) + require.Equal(t, "provider:fixture", rows[0]["source_name"]) + require.Equal(t, int64(5), rows[0]["source_rank"]) + require.Contains(t, rows[0]["normalized_text"], "@U1") + require.Equal(t, "native", rows[1]["text"]) + require.Equal(t, "api-bot", rows[1]["source_name"]) + + catalog, err := st.QueryReadOnly(ctx, ` +select w.name as workspace_name, c.name as channel_name, u.name as user_name +from workspaces w join channels c on c.workspace_id = w.id join users u on u.workspace_id = w.id +where w.id = 'T1' and c.id = 'C1' and u.id = 'U1'`) + require.NoError(t, err) + require.Equal(t, "rich workspace", catalog[0]["workspace_name"]) + require.Equal(t, "rich-channel", catalog[0]["channel_name"]) + require.Equal(t, "rich-user", catalog[0]["user_name"]) +} + +func TestInterruptedFullResumesAndCompletedFullSeedsIncremental(t *testing.T) { + st := openTestStore(t) + ctx := context.Background() + firstRequest := filepath.Join(t.TempDir(), "first.json") + _, err := Sync(ctx, st, helperProvider(t, "fail", firstRequest), Options{WorkspaceID: "T1", Full: true}) + require.Error(t, err) + + fullKey := stateKey("T1", Options{WorkspaceID: "T1", Full: true}) + checkpoint, stateErr := st.GetSyncState(ctx, "provider:fixture", fullKey.entityType, fullKey.entityID) + require.NoError(t, stateErr) + require.Equal(t, `{"cursor":"partial"}`, checkpoint) + + secondRequest := filepath.Join(t.TempDir(), "second.json") + _, err = Sync(ctx, st, helperProvider(t, "success", secondRequest), Options{WorkspaceID: "T1", Full: true}) + require.NoError(t, err) + var sent request + require.NoError(t, json.Unmarshal(requireReadFile(t, secondRequest), &sent)) + require.Equal(t, `{"cursor":"partial"}`, sent.Checkpoint) + + incrementalCheckpoint, stateErr := st.GetSyncState(ctx, "provider:fixture", "workspace", "T1") + require.NoError(t, stateErr) + require.Equal(t, `{"cursor":"done"}`, incrementalCheckpoint) +} + +func TestFailedProcessDoesNotPromoteCompletedFullCheckpoint(t *testing.T) { + st := openTestStore(t) + ctx := context.Background() + require.NoError(t, st.SetSyncState(ctx, "provider:fixture", "workspace", "T1", "old-incremental")) + + _, err := Sync(ctx, st, helperProvider(t, "done-fail", filepath.Join(t.TempDir(), "request.json")), Options{WorkspaceID: "T1", Full: true}) + require.ErrorContains(t, err, "provider exploded after done") + + checkpoint, stateErr := st.GetSyncState(ctx, "provider:fixture", "workspace", "T1") + require.NoError(t, stateErr) + require.Equal(t, "old-incremental", checkpoint) +} + +func TestSyncReturnsProcessErrorAfterSavingAppliedCheckpoint(t *testing.T) { + st := openTestStore(t) + ctx := context.Background() + requestPath := filepath.Join(t.TempDir(), "request.json") + _, err := Sync(ctx, st, helperProvider(t, "fail", requestPath), Options{WorkspaceID: "T1"}) + require.ErrorContains(t, err, "provider exploded") + + checkpoint, stateErr := st.GetSyncState(ctx, "provider:fixture", "workspace", "T1") + require.NoError(t, stateErr) + require.Equal(t, `{"cursor":"partial"}`, checkpoint) + + var sent request + require.NoError(t, json.Unmarshal(requireReadFile(t, requestPath), &sent)) + require.Empty(t, sent.Checkpoint) +} + +func TestSyncRejectsCrossWorkspaceRecord(t *testing.T) { + st := openTestStore(t) + _, err := Sync(context.Background(), st, helperProvider(t, "wrong-workspace", filepath.Join(t.TempDir(), "request.json")), Options{WorkspaceID: "T1"}) + require.ErrorContains(t, err, `belongs to workspace "T2"`) + status, statusErr := st.Status(context.Background()) + require.NoError(t, statusErr) + require.Zero(t, status.Messages) +} + +func TestSyncRejectsProviderIgnoringMessageLimit(t *testing.T) { + st := openTestStore(t) + _, err := Sync(context.Background(), st, helperProvider(t, "too-many-messages", filepath.Join(t.TempDir(), "request.json")), Options{ + WorkspaceID: "T1", + Limit: 2, + }) + require.ErrorContains(t, err, "more than requested message limit 2") + status, statusErr := st.Status(context.Background()) + require.NoError(t, statusErr) + require.Equal(t, 2, status.Messages) +} + +func TestStateKeySeparatesBoundedLimits(t *testing.T) { + limitTwo := stateKey("T1", Options{WorkspaceID: "T1", Limit: 2}) + limitHundred := stateKey("T1", Options{WorkspaceID: "T1", Limit: 100}) + require.NotEqual(t, limitTwo, limitHundred) +} + +func TestSyncRejectsSourceRankTie(t *testing.T) { + st := openTestStore(t) + providerConfig := helperProvider(t, "success", filepath.Join(t.TempDir(), "request.json")) + providerConfig.SourceRank = 2 + _, err := Sync(context.Background(), st, providerConfig, Options{WorkspaceID: "T1"}) + require.ErrorContains(t, err, "greater than 2") +} + +func TestApplyMessageRequiresCatalogReferencesInWorkspace(t *testing.T) { + tests := []struct { + name string + channel *store.Channel + user *store.User + userID string + message string + }{ + {name: "missing channel", message: `missing channel "C1"`}, + {name: "wrong channel workspace", channel: &store.Channel{ID: "C1", WorkspaceID: "T2", Name: "other", Kind: "channel"}, message: `channel "C1" belongs to workspace "T2"`}, + {name: "wrong user workspace", channel: &store.Channel{ID: "C1", WorkspaceID: "T1", Name: "one", Kind: "channel"}, user: &store.User{ID: "U1", WorkspaceID: "T2", Name: "other"}, userID: "U1", message: `user "U1" belongs to workspace "T2"`}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + st := openTestStore(t) + now := time.Now().UTC() + require.NoError(t, st.UpsertWorkspace(context.Background(), store.Workspace{ID: "T1", Name: "one", RawJSON: "{}", UpdatedAt: now})) + require.NoError(t, st.UpsertWorkspace(context.Background(), store.Workspace{ID: "T2", Name: "two", RawJSON: "{}", UpdatedAt: now})) + if test.channel != nil { + test.channel.RawJSON = "{}" + test.channel.UpdatedAt = now + require.NoError(t, st.UpsertChannel(context.Background(), *test.channel)) + } + if test.user != nil { + test.user.RawJSON = "{}" + test.user.UpdatedAt = now + require.NoError(t, st.UpsertUser(context.Background(), *test.user)) + } + consumer := consumer{ + store: st, config: config.Provider{Name: "fixture", SourceRank: 5}, + workspaceID: "T1", sourceName: "provider:fixture", now: now, + validChannels: make(map[string]struct{}), validUsers: make(map[string]struct{}), + } + err := consumer.applyMessage(context.Background(), record{ + WorkspaceID: "T1", ChannelID: "C1", TS: "1.000001", UserID: test.userID, Text: "message", + }) + require.ErrorContains(t, err, test.message) + status, statusErr := st.Status(context.Background()) + require.NoError(t, statusErr) + require.Zero(t, status.Messages) + }) + } +} + +func TestApplyMessagePreservesUncataloguedUserID(t *testing.T) { + st := openTestStore(t) + ctx := context.Background() + now := time.Now().UTC() + require.NoError(t, st.UpsertWorkspace(ctx, store.Workspace{ID: "T1", Name: "one", RawJSON: "{}", UpdatedAt: now})) + require.NoError(t, st.UpsertWorkspace(ctx, store.Workspace{ID: "T2", Name: "two", RawJSON: "{}", UpdatedAt: now})) + require.NoError(t, st.UpsertChannel(ctx, store.Channel{ID: "C1", WorkspaceID: "T1", Name: "one", Kind: "channel", RawJSON: "{}", UpdatedAt: now})) + consumer := consumer{ + store: st, config: config.Provider{Name: "fixture", SourceRank: 5}, + workspaceID: "T1", sourceName: "provider:fixture", now: now, + validChannels: make(map[string]struct{}), validUsers: make(map[string]struct{}), + } + + require.NoError(t, consumer.applyMessage(ctx, record{ + WorkspaceID: "T1", ChannelID: "C1", TS: "1.000001", UserID: "U-missing", Text: "message", + })) + require.NoError(t, consumer.flush(ctx)) + rows, err := st.QueryReadOnly(ctx, `select user_id from messages where channel_id = 'C1' and ts = '1.000001'`) + require.NoError(t, err) + require.Equal(t, "U-missing", rows[0]["user_id"]) + workspaceID, err := st.UserWorkspaceID(ctx, "U-missing") + require.NoError(t, err) + require.Equal(t, "T1", workspaceID) + require.ErrorContains(t, st.EnsureUser(ctx, store.User{ID: "U-missing", WorkspaceID: "T2", Name: "wrong", RawJSON: "{}", UpdatedAt: now}), `belongs to workspace "T1"`) + require.NoError(t, st.EnsureUser(ctx, store.User{ID: "U-missing", WorkspaceID: "T1", Name: "later profile", RawJSON: "{}", UpdatedAt: now})) + workspaceID, err = st.UserWorkspaceID(ctx, "U-missing") + require.NoError(t, err) + require.Equal(t, "T1", workspaceID) + profile, err := st.QueryReadOnly(ctx, `select name from users where id = 'U-missing'`) + require.NoError(t, err) + require.Equal(t, "later profile", profile[0]["name"]) +} + +func TestProviderHelperProcess(t *testing.T) { + separator := -1 + for i, arg := range os.Args { + if arg == "--" { + separator = i + break + } + } + if separator < 0 || len(os.Args) <= separator+2 { + return + } + mode := os.Args[separator+1] + requestPath := os.Args[separator+2] + var sent request + if err := json.NewDecoder(os.Stdin).Decode(&sent); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(2) + } + data, _ := json.Marshal(sent) + _ = os.WriteFile(requestPath, data, 0o600) + emit := func(value any) { + _ = json.NewEncoder(os.Stdout).Encode(value) + } + emit(map[string]any{"type": "hello", "protocol": ProtocolVersion, "provider": "fixture-cache"}) + switch mode { + case "success": + emit(map[string]any{"type": "workspace", "id": "T1", "name": "sparse", "raw_json": map[string]any{}}) + emit(map[string]any{"type": "user", "workspace_id": "T1", "id": "U1", "name": "sparse", "raw_json": map[string]any{}}) + emit(map[string]any{"type": "channel", "workspace_id": "T1", "id": "C1", "name": "sparse", "kind": "channel", "raw_json": map[string]any{}}) + emit(map[string]any{"type": "message", "workspace_id": "T1", "channel_id": "C1", "ts": "1.000001", "user_id": "U1", "text": "provider text <@U1>", "raw_json": map[string]any{}}) + emit(map[string]any{"type": "message", "workspace_id": "T1", "channel_id": "C1", "ts": "2.000002", "user_id": "U1", "text": "lower priority", "raw_json": map[string]any{}}) + emit(map[string]any{"type": "checkpoint", "entity_type": "workspace", "entity_id": "T1", "value": `{"cursor":"done"}`}) + emit(map[string]any{"type": "done", "records": 5}) + os.Exit(0) + case "fail": + emit(map[string]any{"type": "workspace", "id": "T1", "name": "T1", "raw_json": map[string]any{}}) + emit(map[string]any{"type": "checkpoint", "entity_type": "workspace", "entity_id": "T1", "value": `{"cursor":"partial"}`}) + fmt.Fprintln(os.Stderr, "provider exploded") + os.Exit(7) + case "done-fail": + emit(map[string]any{"type": "workspace", "id": "T1", "name": "T1", "raw_json": map[string]any{}}) + emit(map[string]any{"type": "checkpoint", "entity_type": "workspace", "entity_id": "T1", "value": `{"cursor":"done"}`}) + emit(map[string]any{"type": "done", "records": 1}) + fmt.Fprintln(os.Stderr, "provider exploded after done") + os.Exit(7) + case "wrong-workspace": + emit(map[string]any{"type": "message", "workspace_id": "T2", "channel_id": "C1", "ts": "1.000001", "text": "wrong"}) + os.Exit(0) + case "too-many-messages": + emit(map[string]any{"type": "workspace", "id": "T1", "name": "T1", "raw_json": map[string]any{}}) + emit(map[string]any{"type": "user", "workspace_id": "T1", "id": "U1", "name": "one", "raw_json": map[string]any{}}) + emit(map[string]any{"type": "channel", "workspace_id": "T1", "id": "C1", "name": "one", "kind": "channel", "raw_json": map[string]any{}}) + for i := 1; i <= 3; i++ { + emit(map[string]any{"type": "message", "workspace_id": "T1", "channel_id": "C1", "ts": fmt.Sprintf("%d.000001", i), "user_id": "U1", "text": "message"}) + } + emit(map[string]any{"type": "done", "records": 6}) + os.Exit(0) + default: + os.Exit(3) + } +} + +func helperProvider(t *testing.T, mode, requestPath string) config.Provider { + t.Helper() + executable, err := os.Executable() + require.NoError(t, err) + return config.Provider{ + Name: "fixture", Command: executable, + Args: []string{"-test.run=TestProviderHelperProcess", "--", mode, requestPath}, + SourceRank: 5, BatchSize: 7, + } +} + +func openTestStore(t *testing.T) *store.Store { + t.Helper() + st, err := store.Open(filepath.Join(t.TempDir(), "slacrawl.db")) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, st.Close()) }) + return st +} + +func requireReadFile(t *testing.T, path string) []byte { + t.Helper() + data, err := os.ReadFile(path) + require.NoError(t, err) + return data +} diff --git a/internal/share/share.go b/internal/share/share.go index 163980f..cc77293 100644 --- a/internal/share/share.go +++ b/internal/share/share.go @@ -317,6 +317,11 @@ func importLocked(ctx context.Context, s *store.Store, opts Options, restore boo return Manifest{}, err } } + // Keep base rows and their derived search index in one visibility boundary: + // readers see either the previous complete snapshot or the imported one. + if err := store.RebuildSearchIndexesInTransaction(ctx, tx); err != nil { + return Manifest{}, fmt.Errorf("rebuild imported search index: %w", err) + } if err := tx.Commit(); err != nil { return Manifest{}, err } diff --git a/internal/share/share_test.go b/internal/share/share_test.go index 419ee66..cf2d01e 100644 --- a/internal/share/share_test.go +++ b/internal/share/share_test.go @@ -50,6 +50,93 @@ func TestExportImportRoundTrip(t *testing.T) { heads, err := reader.QueryReadOnly(ctx, `select count(*) as count from message_event_heads`) require.NoError(t, err) require.Equal(t, int64(1), heads[0]["count"]) + parity, err := reader.QueryReadOnly(ctx, ` +select count(*) as count +from messages m +join message_fts f + on f.rowid = m.rowid and f.message_key = m.channel_id || '|' || m.ts +`) + require.NoError(t, err) + require.Equal(t, int64(1), parity[0]["count"]) +} + +func TestImportRollsBackWhenAtomicSearchIndexRebuildFails(t *testing.T) { + ctx := context.Background() + dir := t.TempDir() + source := seedStore(t, filepath.Join(dir, "source.db")) + defer func() { require.NoError(t, source.Close()) }() + // Make the snapshot newer than the local reader state so merge import has + // a visible replacement to commit after the forced rollback is removed. + require.NoError(t, source.UpsertMessage(ctx, store.Message{ + ChannelID: "C1", TS: "123.456", WorkspaceID: "T1", UserID: "U1", + Text: "git backed archive works", NormalizedText: "git backed archive works", + SourceRank: 2, SourceName: "api-bot", RawJSON: "{}", UpdatedAt: time.Now().UTC().Add(time.Hour), + }, []store.Mention{{Type: "user", TargetID: "U1", DisplayText: "alice"}})) + // Include a pending marker in the snapshot so the rebuild's final marker + // deletion can be failed deterministically by the reader-side trigger. + require.NoError(t, source.SetSyncState(ctx, "store", "search_index", "rowid_rebuild_pending", "1")) + opts := Options{RepoPath: filepath.Join(dir, "share"), Branch: "main"} + _, err := Export(ctx, source, opts) + require.NoError(t, err) + + readerPath := filepath.Join(dir, "reader.db") + reader := seedStore(t, readerPath) + now := time.Now().UTC() + require.NoError(t, reader.UpsertMessage(ctx, store.Message{ + ChannelID: "C1", TS: "123.456", WorkspaceID: "T1", UserID: "U1", + Text: "reader state before import", NormalizedText: "reader state before import", + SourceRank: 2, SourceName: "api-bot", RawJSON: "{}", UpdatedAt: now, + }, nil)) + _, err = reader.DB().ExecContext(ctx, ` +create trigger reject_search_index_marker_clear +before delete on sync_state +when old.source_name = 'store' + and old.entity_type = 'search_index' + and old.entity_id = 'rowid_rebuild_pending' +begin + select raise(abort, 'forced post-commit rebuild failure'); +end; +`) + require.NoError(t, err) + + _, err = Import(ctx, reader, opts) + require.ErrorContains(t, err, "forced post-commit rebuild failure") + var messages, ftsRows, pending int64 + require.NoError(t, reader.DB().QueryRow(`select count(*) from messages`).Scan(&messages)) + require.NoError(t, reader.DB().QueryRow(`select count(*) from message_fts`).Scan(&ftsRows)) + require.NoError(t, reader.DB().QueryRow(` +select count(*) from sync_state +where source_name = 'store' + and entity_type = 'search_index' + and entity_id = 'rowid_rebuild_pending' +`).Scan(&pending)) + require.Equal(t, int64(1), messages) + require.Equal(t, int64(1), ftsRows) + require.Zero(t, pending) + rows, err := reader.Search(ctx, "", "reader state", 10) + require.NoError(t, err) + require.Len(t, rows, 1) + _, err = reader.DB().ExecContext(ctx, `drop trigger reject_search_index_marker_clear`) + require.NoError(t, err) + + _, err = Import(ctx, reader, opts) + require.NoError(t, err) + rows, err = reader.Search(ctx, "", "archive", 10) + require.NoError(t, err) + require.Len(t, rows, 1) + rows, err = reader.Search(ctx, "", "reader state", 10) + require.NoError(t, err) + require.Empty(t, rows) + require.NoError(t, reader.DB().QueryRow(`select count(*) from message_fts`).Scan(&ftsRows)) + require.NoError(t, reader.DB().QueryRow(` +select count(*) from sync_state +where source_name = 'store' + and entity_type = 'search_index' + and entity_id = 'rowid_rebuild_pending' +`).Scan(&pending)) + require.Equal(t, int64(1), ftsRows) + require.Zero(t, pending) + require.NoError(t, reader.Close()) } func TestImportMergesWithoutRemovingLocalRowsOrNewerTombstones(t *testing.T) { diff --git a/internal/store/batch_test.go b/internal/store/batch_test.go new file mode 100644 index 0000000..328ef7d --- /dev/null +++ b/internal/store/batch_test.go @@ -0,0 +1,389 @@ +package store + +import ( + "context" + "database/sql" + "fmt" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestApplyWriteBatchPreservesHigherPriorityMessageState(t *testing.T) { + st := openBatchTestStore(t) + ctx := context.Background() + now := time.Now().UTC() + seedBatchCatalog(t, st, "T1", "C1", "U1", now) + require.NoError(t, st.UpsertMessage(ctx, Message{ + ChannelID: "C1", TS: "1.000001", WorkspaceID: "T1", UserID: "U1", + Text: "native", NormalizedText: "native", SourceRank: 2, SourceName: "api-bot", + RawJSON: `{"source":"native"}`, UpdatedAt: now, + Files: []MessageFile{{FileID: "F1", Name: "native-file.txt", RawJSON: "{}"}}, + }, []Mention{{Type: "user", TargetID: "U1"}})) + + result, err := st.ApplyWriteBatch(ctx, WriteBatch{ + Messages: []MessageWrite{{ + Message: Message{ + ChannelID: "C1", TS: "1.000001", WorkspaceID: "T1", UserID: "U2", + Text: "provider", NormalizedText: "provider", SourceRank: 5, + SourceName: "provider:test", RawJSON: `{"source":"provider"}`, UpdatedAt: now.Add(time.Second), + }, + Mentions: []Mention{{Type: "user", TargetID: "U2"}}, PreserveHigherPriority: true, + }}, + SyncStates: []SyncStateWrite{{SourceName: "provider:test", EntityType: "workspace", EntityID: "T1", Value: "cursor-1"}}, + }) + require.NoError(t, err) + require.Zero(t, result.MessagesWritten) + + rows, err := st.QueryReadOnly(ctx, `select text, source_name, source_rank from messages where channel_id = 'C1' and ts = '1.000001'`) + require.NoError(t, err) + require.Equal(t, "native", rows[0]["text"]) + require.Equal(t, "api-bot", rows[0]["source_name"]) + require.Equal(t, int64(2), rows[0]["source_rank"]) + assertBatchCount(t, st, `select count(*) from message_files where channel_id = 'C1' and ts = '1.000001' and file_id = 'F1'`, 1) + assertBatchCount(t, st, `select count(*) from message_mentions where channel_id = 'C1' and ts = '1.000001' and target_id = 'U1'`, 1) + assertBatchCount(t, st, `select count(*) from message_mentions where channel_id = 'C1' and ts = '1.000001' and target_id = 'U2'`, 0) + assertBatchCount(t, st, `select count(*) from message_events where channel_id = 'C1' and ts = '1.000001'`, 1) + searchRows, err := st.QueryReadOnly(ctx, `select content from message_fts where message_key = 'C1|1.000001'`) + require.NoError(t, err) + require.Len(t, searchRows, 1) + require.Contains(t, searchRows[0]["content"], "native-file.txt") + checkpoint, err := st.GetSyncState(ctx, "provider:test", "workspace", "T1") + require.NoError(t, err) + require.Equal(t, "cursor-1", checkpoint) +} + +func TestApplyWriteBatchPreservesSequentialMessageSemantics(t *testing.T) { + st := openBatchTestStore(t) + ctx := context.Background() + now := time.Now().UTC() + result, err := st.ApplyWriteBatch(ctx, WriteBatch{ + Workspaces: []Workspace{{ID: "T1", Name: "team", RawJSON: "{}", UpdatedAt: now}}, + Channels: []Channel{{ID: "C1", WorkspaceID: "T1", Name: "general", Kind: "channel", RawJSON: "{}", UpdatedAt: now}}, + Users: []User{{ID: "U1", WorkspaceID: "T1", Name: "one", RawJSON: "{}", UpdatedAt: now}, {ID: "U2", WorkspaceID: "T1", Name: "two", RawJSON: "{}", UpdatedAt: now}}, + Messages: []MessageWrite{ + { + Message: Message{ + ChannelID: "C1", TS: "1.000001", WorkspaceID: "T1", UserID: "U1", + Text: "first", NormalizedText: "first", SourceRank: 5, SourceName: "provider:test", + RawJSON: `{"version":1}`, UpdatedAt: now, + Files: []MessageFile{{FileID: "F1", Name: "retained-file.txt", RawJSON: "{}"}}, + }, + Mentions: []Mention{{Type: "user", TargetID: "U1"}}, PreserveHigherPriority: true, + }, + { + Message: Message{ + ChannelID: "C1", TS: "1.000001", WorkspaceID: "T1", UserID: "U2", + Text: "second", NormalizedText: "second", SourceRank: 5, SourceName: "provider:test", + RawJSON: `{"version":2}`, UpdatedAt: now.Add(time.Second), Files: nil, + }, + Mentions: []Mention{{Type: "user", TargetID: "U2"}}, PreserveHigherPriority: true, + }, + }, + SyncStates: []SyncStateWrite{{SourceName: "provider:test", EntityType: "workspace", EntityID: "T1", Value: "cursor-2"}}, + }) + require.NoError(t, err) + require.Equal(t, 2, result.MessagesWritten) + + rows, err := st.QueryReadOnly(ctx, `select text, user_id, raw_json from messages where channel_id = 'C1' and ts = '1.000001'`) + require.NoError(t, err) + require.Equal(t, "second", rows[0]["text"]) + require.Equal(t, "U2", rows[0]["user_id"]) + require.Equal(t, `{"version":2}`, rows[0]["raw_json"]) + assertBatchCount(t, st, `select count(*) from message_events where channel_id = 'C1' and ts = '1.000001' and source_name = 'provider:test'`, 2) + assertBatchCount(t, st, `select count(*) from message_mentions where channel_id = 'C1' and ts = '1.000001' and target_id = 'U1' and deleted_at is null`, 0) + assertBatchCount(t, st, `select count(*) from message_mentions where channel_id = 'C1' and ts = '1.000001' and target_id = 'U2' and deleted_at is null`, 1) + assertBatchCount(t, st, `select count(*) from message_files where channel_id = 'C1' and ts = '1.000001' and file_id = 'F1'`, 1) + searchRows, err := st.QueryReadOnly(ctx, `select content from message_fts where message_key = 'C1|1.000001'`) + require.NoError(t, err) + require.Len(t, searchRows, 1) + require.Contains(t, searchRows[0]["content"], "retained-file.txt") +} + +func TestApplyWriteBatchRollsBackMessagesAndCheckpointOnFailure(t *testing.T) { + st := openBatchTestStore(t) + ctx := context.Background() + now := time.Now().UTC() + seedBatchCatalog(t, st, "T1", "C1", "U1", now) + require.NoError(t, st.UpsertWorkspace(ctx, Workspace{ID: "T2", Name: "two", RawJSON: "{}", UpdatedAt: now})) + require.NoError(t, st.UpsertChannel(ctx, Channel{ID: "C2", WorkspaceID: "T2", Name: "two", Kind: "channel", RawJSON: "{}", UpdatedAt: now})) + require.NoError(t, st.UpsertMessage(ctx, Message{ + ChannelID: "C2", TS: "2.000002", WorkspaceID: "T2", Text: "existing", + NormalizedText: "existing", SourceRank: 2, SourceName: "api-bot", RawJSON: "{}", UpdatedAt: now, + }, nil)) + + _, err := st.ApplyWriteBatch(ctx, WriteBatch{ + Messages: []MessageWrite{ + {Message: batchMessage("C1", "1.000001", "T1", "first", now), PreserveHigherPriority: true}, + {Message: batchMessage("C2", "2.000002", "T1", "collision", now), PreserveHigherPriority: true}, + }, + SyncStates: []SyncStateWrite{{SourceName: "provider:test", EntityType: "workspace", EntityID: "T1", Value: "must-not-commit"}}, + }) + require.ErrorContains(t, err, "already belongs to workspace") + assertBatchCount(t, st, `select count(*) from messages where channel_id = 'C1' and ts = '1.000001'`, 0) + assertBatchCount(t, st, `select count(*) from sync_state where source_name = 'provider:test'`, 0) +} + +func TestApplyWriteBatchRollsBackWhenCheckpointWriteFails(t *testing.T) { + st := openBatchTestStore(t) + ctx := context.Background() + now := time.Now().UTC() + seedBatchCatalog(t, st, "T1", "C1", "U1", now) + _, err := st.DB().ExecContext(ctx, ` +create trigger reject_batch_checkpoint before insert on sync_state +when new.source_name = 'provider:test' +begin select raise(abort, 'checkpoint rejected'); end`) + require.NoError(t, err) + + _, err = st.ApplyWriteBatch(ctx, WriteBatch{ + Messages: []MessageWrite{{Message: batchMessage("C1", "1.000001", "T1", "message", now), PreserveHigherPriority: true}}, + SyncStates: []SyncStateWrite{{SourceName: "provider:test", EntityType: "workspace", EntityID: "T1", Value: "cursor"}}, + }) + require.ErrorContains(t, err, "checkpoint rejected") + assertBatchCount(t, st, `select count(*) from messages where channel_id = 'C1' and ts = '1.000001'`, 0) + assertBatchCount(t, st, `select count(*) from message_fts where message_key = 'C1|1.000001'`, 0) + assertBatchCount(t, st, `select count(*) from message_events where channel_id = 'C1' and ts = '1.000001'`, 0) +} + +func TestApplyWriteBatchHonorsPerMessageRetention(t *testing.T) { + st := openBatchTestStore(t) + ctx := context.Background() + now := time.Now().UTC() + seedBatchCatalog(t, st, "T1", "C1", "U1", now) + require.NoError(t, st.UpsertMessage(ctx, batchMessage("C1", "50.000001", "T1", "existing old", now), nil)) + require.NoError(t, st.SetSyncState(ctx, retentionFloorSource, retentionFloorEntityType, "T1|C1", "100.000000")) + + result, err := st.ApplyWriteBatch(ctx, WriteBatch{Messages: []MessageWrite{ + {Message: batchMessage("C1", "50.000001", "T1", "updated old", now.Add(time.Second)), PreserveHigherPriority: true, EnforceRetention: true}, + {Message: batchMessage("C1", "60.000001", "T1", "blocked old", now), PreserveHigherPriority: true, EnforceRetention: true}, + {Message: batchMessage("C1", "70.000001", "T1", "explicit restore", now), PreserveHigherPriority: true, EnforceRetention: false}, + {Message: batchMessage("C1", "150.000001", "T1", "new", now), PreserveHigherPriority: true, EnforceRetention: true}, + }}) + require.NoError(t, err) + require.Equal(t, 3, result.MessagesWritten) + assertBatchCount(t, st, `select count(*) from messages where channel_id = 'C1' and ts = '60.000001'`, 0) + assertBatchCount(t, st, `select count(*) from messages where channel_id = 'C1' and ts in ('50.000001', '70.000001', '150.000001')`, 3) +} + +func TestWriteMessageFTSRequiresCanonicalMessageRow(t *testing.T) { + ctx := context.Background() + writer := &recordingFTSWriter{rows: 1} + require.NoError(t, writeMessageFTS(ctx, writer, "C1", "1.000001", "new")) + require.Equal(t, []string{upsertMessageFTSRowSQL}, writer.queries) + require.Equal(t, [][]any{{"new", "C1", "1.000001"}}, writer.args) + + writer.rows = 0 + require.ErrorContains(t, writeMessageFTS(ctx, writer, "C2", "2.000002", "missing"), "affected 0 rows") +} + +func TestApplyWriteBatchFTSInsertAndUpdateParity(t *testing.T) { + st := openBatchTestStore(t) + ctx := context.Background() + now := time.Now().UTC() + seedBatchCatalog(t, st, "T1", "C1", "U1", now) + require.NoError(t, st.UpsertMessage(ctx, batchMessage("C1", "1.000001", "T1", "unrelated", now), nil)) + + insertResult, err := st.ApplyWriteBatch(ctx, WriteBatch{Messages: []MessageWrite{{ + Message: batchMessage("C1", "2.000002", "T1", "inserted", now), PreserveHigherPriority: true, + }}}) + require.NoError(t, err) + require.Equal(t, 1, insertResult.MessagesWritten) + assertBatchFTSContent(t, st, "C1|1.000001", "unrelated") + assertBatchFTSContent(t, st, "C1|2.000002", "inserted") + assertBatchCount(t, st, `select count(*) from message_fts where message_key = 'C1|2.000002'`, 1) + + updateResult, err := st.ApplyWriteBatch(ctx, WriteBatch{Messages: []MessageWrite{{ + Message: batchMessage("C1", "2.000002", "T1", "updated", now.Add(time.Second)), PreserveHigherPriority: true, + }}}) + require.NoError(t, err) + require.Equal(t, 1, updateResult.MessagesWritten) + assertBatchFTSContent(t, st, "C1|1.000001", "unrelated") + assertBatchFTSContent(t, st, "C1|2.000002", "updated") + assertBatchCount(t, st, `select count(*) from message_fts where message_key = 'C1|2.000002'`, 1) +} + +func TestApplyWriteBatchProviderFastPathChecksRowsAndMentions(t *testing.T) { + st := openBatchTestStore(t) + ctx := context.Background() + now := time.Now().UTC() + seedBatchCatalog(t, st, "T1", "C1", "U1", now) + message := batchMessage("C1", "1.000001", "T1", "original", now) + require.NoError(t, st.UpsertMessage(ctx, message, nil)) + + unchanged := message + unchanged.UpdatedAt = now.Add(time.Second) + result, err := st.ApplyWriteBatch(ctx, WriteBatch{Messages: []MessageWrite{{ + Message: unchanged, PreserveHigherPriority: true, SkipUnchangedProviderRow: true, + }}}) + require.NoError(t, err) + require.Zero(t, result.MessagesWritten) + assertBatchCount(t, st, `select count(*) from message_events where channel_id = 'C1' and ts = '1.000001'`, 1) + + edited := unchanged + edited.Text = "edited" + edited.NormalizedText = "edited" + result, err = st.ApplyWriteBatch(ctx, WriteBatch{Messages: []MessageWrite{{ + Message: edited, PreserveHigherPriority: true, SkipUnchangedProviderRow: true, + }}}) + require.NoError(t, err) + require.Equal(t, 1, result.MessagesWritten) + assertBatchFTSContent(t, st, "C1|1.000001", "edited") + assertBatchCount(t, st, `select count(*) from message_events where channel_id = 'C1' and ts = '1.000001'`, 1) + + result, err = st.ApplyWriteBatch(ctx, WriteBatch{Messages: []MessageWrite{{ + Message: edited, Mentions: []Mention{{Type: "user", TargetID: "U2"}}, + PreserveHigherPriority: true, SkipUnchangedProviderRow: true, + }}}) + require.NoError(t, err) + require.Equal(t, 1, result.MessagesWritten) + assertBatchCount(t, st, `select count(*) from message_mentions where channel_id = 'C1' and ts = '1.000001' and target_id = 'U2'`, 1) +} + +func TestApplyWriteBatchProviderFastPathBatchesLargeReplays(t *testing.T) { + st := openBatchTestStore(t) + ctx := context.Background() + now := time.Now().UTC() + seedBatchCatalog(t, st, "T1", "C1", "U1", now) + require.NoError(t, st.UpsertUser(ctx, User{ID: "U2", WorkspaceID: "T1", Name: "two", RawJSON: "{}", UpdatedAt: now})) + + const unchangedCandidates = 1001 + seedWrites := make([]MessageWrite, 0, unchangedCandidates+1) + for i := range unchangedCandidates { + message := batchMessage("C1", fmt.Sprintf("%d.000001", i+1), "T1", fmt.Sprintf("message-%d", i), now) + var mentions []Mention + if i == 1 { + mentions = []Mention{{Type: "user", TargetID: "U1", DisplayText: "one"}} + } + seedWrites = append(seedWrites, MessageWrite{Message: message, Mentions: mentions, PreserveHigherPriority: true}) + } + duplicate := batchMessage("C1", "2000.000001", "T1", "duplicate-original", now) + seedWrites = append(seedWrites, MessageWrite{Message: duplicate, PreserveHigherPriority: true}) + seedResult, err := st.ApplyWriteBatch(ctx, WriteBatch{Messages: seedWrites}) + require.NoError(t, err) + require.Equal(t, unchangedCandidates+1, seedResult.MessagesWritten) + requireMessageFTSParity(t, st) + + _, err = st.DB().ExecContext(ctx, ` +update message_event_heads set payload_json = '{"stale":true}' +where channel_id = 'C1' and ts = '3.000001'; +create table batch_message_updates (count integer not null); +insert into batch_message_updates values (0); +create trigger count_batch_message_updates after update on messages +begin + update batch_message_updates set count = count + 1; +end; +`) + require.NoError(t, err) + + replayWrites := make([]MessageWrite, 0, unchangedCandidates+2) + for i := range unchangedCandidates { + message := batchMessage("C1", fmt.Sprintf("%d.000001", i+1), "T1", fmt.Sprintf("message-%d", i), now.Add(time.Second)) + var mentions []Mention + switch i { + case 0: + message.Text = "changed-message" + message.NormalizedText = "changed-message" + message.RawJSON = `{"text":"changed-message"}` + case 1: + mentions = []Mention{{Type: "user", TargetID: "U2", DisplayText: "two"}} + } + replayWrites = append(replayWrites, MessageWrite{ + Message: message, Mentions: mentions, PreserveHigherPriority: true, SkipUnchangedProviderRow: true, + }) + } + duplicateFirst := duplicate + duplicateFirst.Text = "duplicate-first" + duplicateFirst.NormalizedText = "duplicate-first" + duplicateFirst.RawJSON = `{"text":"duplicate-first"}` + duplicateSecond := duplicateFirst + duplicateSecond.Text = "duplicate-second" + duplicateSecond.NormalizedText = "duplicate-second" + duplicateSecond.RawJSON = `{"text":"duplicate-second"}` + replayWrites = append(replayWrites, + MessageWrite{Message: duplicateFirst, PreserveHigherPriority: true, SkipUnchangedProviderRow: true}, + MessageWrite{Message: duplicateSecond, PreserveHigherPriority: true, SkipUnchangedProviderRow: true}, + ) + + result, err := st.ApplyWriteBatch(ctx, WriteBatch{Messages: replayWrites}) + require.NoError(t, err) + require.Equal(t, 5, result.MessagesWritten) + assertBatchCount(t, st, `select count(*) from batch_message_updates where count = 5`, 1) + assertBatchFTSContent(t, st, "C1|1.000001", "changed-message") + assertBatchCount(t, st, `select count(*) from message_mentions where channel_id = 'C1' and ts = '2.000001' and target_id = 'U2' and display_text = 'two'`, 1) + assertBatchCount(t, st, `select count(*) from message_event_heads where channel_id = 'C1' and ts = '3.000001' and payload_json = '{"text":"message-2"}'`, 1) + assertBatchFTSContent(t, st, "C1|2000.000001", "duplicate-second") + requireMessageFTSParity(t, st) +} + +type recordingFTSWriter struct { + rows int64 + queries []string + args [][]any +} + +func (w *recordingFTSWriter) ExecContext(_ context.Context, query string, args ...any) (sql.Result, error) { + w.queries = append(w.queries, query) + w.args = append(w.args, args) + return recordingSQLResult(w.rows), nil +} + +type recordingSQLResult int64 + +func (r recordingSQLResult) LastInsertId() (int64, error) { return 0, nil } +func (r recordingSQLResult) RowsAffected() (int64, error) { return int64(r), nil } + +func openBatchTestStore(t *testing.T) *Store { + t.Helper() + st, err := Open(filepath.Join(t.TempDir(), "batch.db")) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, st.Close()) }) + return st +} + +func seedBatchCatalog(t *testing.T, st *Store, workspaceID, channelID, userID string, now time.Time) { + t.Helper() + require.NoError(t, st.UpsertWorkspace(context.Background(), Workspace{ID: workspaceID, Name: workspaceID, RawJSON: "{}", UpdatedAt: now})) + require.NoError(t, st.UpsertChannel(context.Background(), Channel{ID: channelID, WorkspaceID: workspaceID, Name: channelID, Kind: "channel", RawJSON: "{}", UpdatedAt: now})) + require.NoError(t, st.UpsertUser(context.Background(), User{ID: userID, WorkspaceID: workspaceID, Name: userID, RawJSON: "{}", UpdatedAt: now})) +} + +func batchMessage(channelID, ts, workspaceID, text string, now time.Time) Message { + return Message{ + ChannelID: channelID, TS: ts, WorkspaceID: workspaceID, UserID: "U1", + Text: text, NormalizedText: text, SourceRank: 5, SourceName: "provider:test", + RawJSON: `{"text":"` + text + `"}`, UpdatedAt: now, + } +} + +func assertBatchCount(t *testing.T, st *Store, query string, expected int64) { + t.Helper() + rows, err := st.QueryReadOnly(context.Background(), query) + require.NoError(t, err) + require.Len(t, rows, 1) + require.Equal(t, expected, rows[0]["count(*)"]) +} + +func assertBatchFTSContent(t *testing.T, st *Store, key, expected string) { + t.Helper() + rows, err := st.QueryReadOnly(context.Background(), `select content from message_fts where message_key = '`+key+`'`) + require.NoError(t, err) + require.Len(t, rows, 1) + require.Equal(t, expected, rows[0]["content"]) +} + +func requireMessageFTSParity(t *testing.T, st *Store) { + t.Helper() + var messages, ftsRows, matched int64 + require.NoError(t, st.DB().QueryRow(`select count(*) from messages`).Scan(&messages)) + require.NoError(t, st.DB().QueryRow(`select count(*) from message_fts`).Scan(&ftsRows)) + require.NoError(t, st.DB().QueryRow(` +select count(*) +from messages m +join message_fts f + on f.rowid = m.rowid and f.message_key = m.channel_id || '|' || m.ts +`).Scan(&matched)) + require.Equal(t, messages, ftsRows) + require.Equal(t, messages, matched) +} diff --git a/internal/store/purge.go b/internal/store/purge.go index d940cc9..7dc275a 100644 --- a/internal/store/purge.go +++ b/internal/store/purge.go @@ -3,6 +3,7 @@ package store import ( "context" "database/sql" + "database/sql/driver" "errors" "fmt" "strings" @@ -381,7 +382,9 @@ join messages m on m.workspace_id = p.workspace_id and m.channel_id = e.channel_ {"FTS entries", ` select count(*) from message_fts f -join ` + purgeMessageKeysTable + ` p on f.message_key = p.channel_id || '|' || p.ts +join messages m on m.rowid = f.rowid +join ` + purgeMessageKeysTable + ` p + on p.workspace_id = m.workspace_id and p.channel_id = m.channel_id and p.ts = m.ts `, &report.FTSEntries}, } for _, count := range counts { @@ -471,9 +474,11 @@ where exists ( )`}, {"FTS entries", ` delete from message_fts -where exists ( - select 1 from ` + purgeMessageKeysTable + ` p - where message_fts.message_key = p.channel_id || '|' || p.ts +where rowid in ( + select m.rowid + from messages m + join ` + purgeMessageKeysTable + ` p + on p.workspace_id = m.workspace_id and p.channel_id = m.channel_id and p.ts = m.ts )`}, {"messages", ` delete from messages @@ -504,8 +509,126 @@ where id in (select id from `+purgeMessageEventIDsTable+`) } func (s *Store) Vacuum(ctx context.Context) error { - _, err := s.db.ExecContext(ctx, "vacuum") - return err + return s.vacuum(ctx, nil) +} + +func (s *Store) vacuum(ctx context.Context, afterPrepare func()) error { + conn, err := s.db.Conn(ctx) + if err != nil { + return err + } + var databaseSequence int + var databaseName, databasePath string + if err := conn.QueryRowContext(ctx, `pragma database_list`).Scan(&databaseSequence, &databaseName, &databasePath); err != nil { + _ = conn.Close() + return fmt.Errorf("inspect vacuum database: %w", err) + } + if databaseName != "main" || strings.TrimSpace(databasePath) == "" { + _ = conn.Close() + return errors.New("vacuum is not supported for in-memory databases") + } + poisonStore := false + defer releaseExclusiveSQLiteConnection(conn) + defer func() { + if poisonStore { + _ = s.db.Close() + return + } + s.searchIndexUnavailable.Store(false) + }() + + var lockingMode string + if err := conn.QueryRowContext(ctx, `pragma locking_mode = exclusive`).Scan(&lockingMode); err != nil { + return fmt.Errorf("enable exclusive vacuum locking: %w", err) + } + if !strings.EqualFold(lockingMode, "exclusive") { + return fmt.Errorf("enable exclusive vacuum locking: sqlite returned %q", lockingMode) + } + s.searchIndexUnavailable.Store(true) + if _, err := conn.ExecContext(ctx, `begin exclusive`); err != nil { + return fmt.Errorf("lock database for vacuum: %w", err) + } + if err := markSearchIndexRebuildPending(ctx, conn); err != nil { + return fmt.Errorf("mark search index for post-vacuum rebuild: %w", err) + } + if _, err := conn.ExecContext(ctx, `delete from message_fts`); err != nil { + return fmt.Errorf("clear search index before vacuum: %w", err) + } + if _, err := conn.ExecContext(ctx, `commit`); err != nil { + recoveryErr := recoverSearchIndexOnExclusiveConnection(conn) + if recoveryErr != nil { + poisonStore = true + } + return errors.Join( + fmt.Errorf("prepare search index for vacuum: %w", err), + wrapSearchIndexRecoveryError(recoveryErr), + ) + } + if afterPrepare != nil { + afterPrepare() + } + + _, vacuumErr := conn.ExecContext(ctx, "vacuum") + if rebuildErr := rebuildSearchIndexesOnExclusiveConnection(ctx, conn); rebuildErr != nil { + recoveryErr := recoverSearchIndexOnExclusiveConnection(conn) + if recoveryErr != nil { + poisonStore = true + } + return errors.Join( + vacuumErr, + fmt.Errorf("rebuild search index after vacuum: %w", rebuildErr), + wrapSearchIndexRecoveryError(recoveryErr), + ) + } + return vacuumErr +} + +func recoverSearchIndexOnExclusiveConnection(conn *sql.Conn) error { + recoveryCtx, cancel := context.WithTimeout(context.Background(), 10*time.Minute) + defer cancel() + _, _ = conn.ExecContext(recoveryCtx, `rollback`) + return rebuildSearchIndexesOnExclusiveConnection(recoveryCtx, conn) +} + +func wrapSearchIndexRecoveryError(err error) error { + if err == nil { + return nil + } + return fmt.Errorf("recover search index after vacuum failure: %w", err) +} + +func rebuildSearchIndexesOnExclusiveConnection(ctx context.Context, conn *sql.Conn) error { + if _, err := conn.ExecContext(ctx, `begin exclusive`); err != nil { + return err + } + if err := rebuildSearchIndexesInTransaction(ctx, conn); err != nil { + _, _ = conn.ExecContext(context.Background(), `rollback`) + return err + } + if _, err := conn.ExecContext(ctx, `commit`); err != nil { + _, _ = conn.ExecContext(context.Background(), `rollback`) + return err + } + return nil +} + +func releaseExclusiveSQLiteConnection(conn *sql.Conn) { + cleanupCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _, _ = conn.ExecContext(cleanupCtx, `rollback`) + var lockingMode string + err := conn.QueryRowContext(cleanupCtx, `pragma locking_mode = normal`).Scan(&lockingMode) + if err == nil && !strings.EqualFold(lockingMode, "normal") { + err = fmt.Errorf("sqlite returned locking mode %q", lockingMode) + } + if err == nil { + var schemaVersion int + err = conn.QueryRowContext(cleanupCtx, `pragma user_version`).Scan(&schemaVersion) + } + if err != nil { + _ = conn.Raw(func(any) error { return driver.ErrBadConn }) + } + _ = conn.Close() } func (s *Store) UnreferencedPurgeMediaAfter(ctx context.Context, items, removed []PurgeMedia) ([]PurgeMedia, int64, error) { diff --git a/internal/store/purge_test.go b/internal/store/purge_test.go index 40e98c5..4c8701f 100644 --- a/internal/store/purge_test.go +++ b/internal/store/purge_test.go @@ -2,6 +2,7 @@ package store import ( "context" + "database/sql" "fmt" "path/filepath" "testing" @@ -26,6 +27,7 @@ func TestPurgeMessagesPreviewsAndDeletesMessageOwnedRows(t *testing.T) { upsertPurgeTestMessage(t, st, "T1", "C1", oldTime.Add(time.Second), "old shared", "F2", sharedMedia, 20) upsertPurgeTestMessage(t, st, "T1", "C1", newTime, "new shared", "F3", sharedMedia, 20) upsertPurgeTestMessage(t, st, "T2", "C2", oldTime, "other workspace", "F4", "files/cc/other.txt", 30) + requireMessageFTSParity(t, st) _, err = st.DB().ExecContext(ctx, ` insert into embedding_jobs (channel_id, ts, state, created_at) @@ -56,6 +58,7 @@ values (?, ?, 'pending', ?), (?, ?, 'pending', ?) requireTableCount(t, st, "message_mentions", 2) requireTableCount(t, st, "embedding_jobs", 1) requireTableCount(t, st, "message_fts", 2) + requireMessageFTSParity(t, st) rows, err := st.QueryReadOnly(ctx, `select text from messages order by text`) require.NoError(t, err) @@ -70,6 +73,215 @@ values (?, ?, 'pending', ?), (?, ?, 'pending', ?) require.Empty(t, empty.Media) } +func TestVacuumRebuildsMessageFTSRowIDs(t *testing.T) { + st, err := Open(filepath.Join(t.TempDir(), "slacrawl.db")) + require.NoError(t, err) + defer func() { require.NoError(t, st.Close()) }() + ctx := context.Background() + now := time.Now().UTC() + upsertPurgeTestMessage(t, st, "T1", "C1", now, "first vacuum message", "F1", "", 0) + upsertPurgeTestMessage(t, st, "T1", "C1", now.Add(time.Second), "second vacuum message", "F2", "", 0) + + _, err = st.DB().ExecContext(ctx, ` +delete from message_fts; +insert into message_fts (rowid, message_key, content) values + (1001, 'C1|wrong-one', 'wrong one'), + (1002, 'C1|wrong-two', 'wrong two'); +`) + require.NoError(t, err) + require.NoError(t, st.Vacuum(ctx)) + requireMessageFTSParity(t, st) + matches, err := st.Search(ctx, "T1", "vacuum", 10) + require.NoError(t, err) + require.Len(t, matches, 2) + assertBatchCount(t, st, `select count(*) from sync_state where source_name = 'store' and entity_type = 'search_index'`, 0) +} + +func TestVacuumHoldsExclusiveLockAcrossRebuild(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "slacrawl.db") + st, err := Open(dbPath) + require.NoError(t, err) + defer func() { require.NoError(t, st.Close()) }() + ctx := context.Background() + upsertPurgeTestMessage(t, st, "T1", "C1", time.Now().UTC(), "exclusive vacuum", "F1", "", 0) + + prepared := make(chan struct{}) + release := make(chan struct{}) + vacuumDone := make(chan error, 1) + go func() { + vacuumDone <- st.vacuum(ctx, func() { + close(prepared) + <-release + }) + }() + select { + case <-prepared: + case <-time.After(5 * time.Second): + close(release) + t.Fatal("vacuum did not reach prepared state") + } + + type openResult struct { + store *Store + err error + } + openStarted := make(chan struct{}) + openDone := make(chan openResult, 1) + go func() { + close(openStarted) + opened, err := Open(dbPath) + openDone <- openResult{store: opened, err: err} + }() + <-openStarted + select { + case opened := <-openDone: + close(release) + if opened.store != nil { + _ = opened.store.Close() + } + t.Fatalf("concurrent open completed while vacuum held its exclusive lock: %v", opened.err) + case <-time.After(200 * time.Millisecond): + } + close(release) + select { + case err := <-vacuumDone: + require.NoError(t, err) + case <-time.After(10 * time.Second): + t.Fatal("vacuum did not finish after release") + } + select { + case opened := <-openDone: + require.NoError(t, opened.err) + require.NoError(t, opened.store.Close()) + case <-time.After(10 * time.Second): + t.Fatal("concurrent open did not finish after vacuum released its lock") + } + requireMessageFTSParity(t, st) +} + +func TestVacuumRejectsInMemoryDatabaseWithoutDamagingIndex(t *testing.T) { + st, err := Open(":memory:") + require.NoError(t, err) + defer func() { require.NoError(t, st.Close()) }() + ctx := context.Background() + message := Message{ + ChannelID: "C1", TS: "1.000001", WorkspaceID: "T1", + Text: "in-memory vacuum guard", NormalizedText: "in-memory vacuum guard", + SourceRank: 2, SourceName: "test", RawJSON: "{}", UpdatedAt: time.Now().UTC(), + } + require.NoError(t, st.UpsertMessage(ctx, message, nil)) + require.ErrorContains(t, st.Vacuum(ctx), "not supported for in-memory databases") + requireMessageFTSParity(t, st) + matches, err := st.Search(ctx, "", "vacuum guard", 10) + require.NoError(t, err) + require.Len(t, matches, 1) +} + +func TestVacuumCancellationRecoversIndexForSameStore(t *testing.T) { + st, err := Open(filepath.Join(t.TempDir(), "slacrawl.db")) + require.NoError(t, err) + defer func() { require.NoError(t, st.Close()) }() + message := Message{ + ChannelID: "C1", TS: "1.000001", WorkspaceID: "T1", + Text: "vacuum cancellation recovery", NormalizedText: "vacuum cancellation recovery", + SourceRank: 2, SourceName: "test", RawJSON: "{}", UpdatedAt: time.Now().UTC(), + } + require.NoError(t, st.UpsertMessage(context.Background(), message, nil)) + ctx, cancel := context.WithCancel(context.Background()) + err = st.vacuum(ctx, cancel) + require.ErrorIs(t, err, context.Canceled) + requireMessageFTSParity(t, st) + matches, err := st.Search(context.Background(), "", "cancellation recovery", 10) + require.NoError(t, err) + require.Len(t, matches, 1) + assertBatchCount(t, st, `select count(*) from sync_state where source_name = 'store' and entity_type = 'search_index'`, 0) +} + +func TestVacuumUnrecoverableRebuildClosesStore(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "slacrawl.db") + st, err := Open(dbPath) + require.NoError(t, err) + message := Message{ + ChannelID: "C1", TS: "1.000001", WorkspaceID: "T1", + Text: "vacuum poison recovery", NormalizedText: "vacuum poison recovery", + SourceRank: 2, SourceName: "test", RawJSON: "{}", UpdatedAt: time.Now().UTC(), + } + require.NoError(t, st.UpsertMessage(context.Background(), message, nil)) + _, err = st.DB().Exec(` +create trigger reject_vacuum_marker_clear +before delete on sync_state +when old.source_name = 'store' + and old.entity_type = 'search_index' + and old.entity_id = 'rowid_rebuild_pending' +begin + select raise(abort, 'forced vacuum rebuild failure'); +end; +`) + require.NoError(t, err) + + type searchResult struct { + rows []MessageRow + err error + } + waiterStarted := make(chan struct{}) + waiterDone := make(chan searchResult, 1) + err = st.vacuum(context.Background(), func() { + go func() { + close(waiterStarted) + rows, err := st.Search(context.Background(), "", "poison", 10) + waiterDone <- searchResult{rows: rows, err: err} + }() + <-waiterStarted + }) + require.ErrorContains(t, err, "forced vacuum rebuild failure") + waiter := <-waiterDone + require.Error(t, waiter.err) + require.Empty(t, waiter.rows) + _, err = st.Search(context.Background(), "", "poison", 10) + require.ErrorContains(t, err, "search index is temporarily unavailable") + require.ErrorContains(t, st.DB().Ping(), "closed") + + db, err := sql.Open("sqlite", dbPath) + require.NoError(t, err) + _, err = db.Exec(`drop trigger reject_vacuum_marker_clear`) + require.NoError(t, err) + require.NoError(t, db.Close()) + + repaired, err := Open(dbPath) + require.NoError(t, err) + defer func() { require.NoError(t, repaired.Close()) }() + requireMessageFTSParity(t, repaired) + matches, err := repaired.Search(context.Background(), "", "poison recovery", 10) + require.NoError(t, err) + require.Len(t, matches, 1) +} + +func TestOpenRepairsInterruptedVacuumSearchIndex(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "slacrawl.db") + st, err := Open(dbPath) + require.NoError(t, err) + ctx := context.Background() + now := time.Now().UTC() + upsertPurgeTestMessage(t, st, "T1", "C1", now, "repair after interruption", "F1", "", 0) + require.NoError(t, st.SetSyncState(ctx, searchIndexMaintenanceSource, searchIndexMaintenanceEntityType, searchIndexMaintenanceEntityID, "1")) + _, err = st.DB().ExecContext(ctx, `delete from message_fts`) + require.NoError(t, err) + require.NoError(t, st.Close()) + + readOnly, err := OpenReadOnly(dbPath) + require.ErrorContains(t, err, "search index rebuild is pending") + require.Nil(t, readOnly) + + st, err = Open(dbPath) + require.NoError(t, err) + defer func() { require.NoError(t, st.Close()) }() + requireMessageFTSParity(t, st) + matches, err := st.Search(ctx, "T1", "interruption", 10) + require.NoError(t, err) + require.Len(t, matches, 1) + assertBatchCount(t, st, `select count(*) from sync_state where source_name = 'store' and entity_type = 'search_index'`, 0) +} + func TestPurgeMessagesCompactsRetainedEventHistoryBySourceAndType(t *testing.T) { st, err := Open(filepath.Join(t.TempDir(), "slacrawl.db")) require.NoError(t, err) diff --git a/internal/store/sqlc/queries.sql b/internal/store/sqlc/queries.sql index b8c6167..d2dc80e 100644 --- a/internal/store/sqlc/queries.sql +++ b/internal/store/sqlc/queries.sql @@ -186,12 +186,6 @@ set fetched_at = ?, updated_at = ? where channel_id = ? and ts = ? and file_id = ?; --- name: DeleteMessageFTS :exec -delete from message_fts where message_key = ?; - --- name: InsertMessageFTS :exec -insert into message_fts (message_key, content) values (?, ?); - -- name: InsertMessageEvent :exec insert into message_events (channel_id, ts, event_type, source_name, payload_json, created_at) values (?, ?, ?, ?, ?, ?); diff --git a/internal/store/store.go b/internal/store/store.go index 06e9e76..fe60ac8 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -12,6 +12,7 @@ import ( "sort" "strconv" "strings" + "sync/atomic" "time" "unicode" @@ -20,6 +21,13 @@ import ( ) const schemaVersion = 6 +const PlaceholderUserRawJSON = `{"slacrawl_provider_placeholder":true}` + +const ( + searchIndexMaintenanceSource = "store" + searchIndexMaintenanceEntityType = "search_index" + searchIndexMaintenanceEntityID = "rowid_rebuild_pending" +) const schemaPragmas = ` pragma foreign_keys = on; @@ -261,7 +269,7 @@ create index if not exists idx_message_files_name on message_files(name); const schemaV4Migration = messageEventHeadsSchema const schemaV5Migration = messageEventHeadTriggerSchema -const schemaV6Migration = ` +const schemaV6EventMigration = ` drop index if exists idx_message_events_identity; create table message_events_v6 ( id integer primary key autoincrement, @@ -282,9 +290,35 @@ create unique index if not exists idx_message_events_identity on message_events(event_key); ` +const rebuildMessageFTSRowsSQL = ` +insert into message_fts (rowid, message_key, content) +select m.rowid, + m.channel_id || '|' || m.ts, + trim(m.normalized_text || ' ' || coalesce(( + select group_concat(trim(f.name || ' ' || f.title || ' ' || f.plain_text || ' ' || f.preview_plain_text), ' ') + from message_files f + where f.channel_id = m.channel_id and f.ts = m.ts and f.deleted_at is null + ), '')) +from messages m +` + +const upsertMessageFTSRowSQL = ` +insert or replace into message_fts (rowid, message_key, content) +select rowid, channel_id || '|' || ts, ? +from messages +where channel_id = ? and ts = ? +` + +const deleteMessageFTSRowSQL = ` +delete from message_fts +where rowid = (select rowid from messages where channel_id = ? and ts = ?) +` + type Store struct { - db *sql.DB - q *storedb.Queries + db *sql.DB + q *storedb.Queries + ftsRowIDAligned bool + searchIndexUnavailable atomic.Bool } type Workspace struct { @@ -346,6 +380,36 @@ type Message struct { Files []MessageFile } +type MessageWrite struct { + Message Message + Mentions []Mention + PreserveHigherPriority bool + EnforceRetention bool + // SkipUnchangedProviderRow avoids derived-index rewrites for provider rows + // whose scalar state, mentions, and event head already match. It is not a + // general repair path and deliberately refuses messages with file payloads. + SkipUnchangedProviderRow bool +} + +type SyncStateWrite struct { + SourceName string + EntityType string + EntityID string + Value string +} + +type WriteBatch struct { + Workspaces []Workspace + Channels []Channel + Users []User + Messages []MessageWrite + SyncStates []SyncStateWrite +} + +type WriteBatchResult struct { + MessagesWritten int +} + type Mention struct { Type string TargetID string @@ -631,7 +695,22 @@ func Open(path string) (*Store, error) { return nil, err } } - return &Store{db: db, q: storedb.New(db)}, nil + if err := repairPendingSearchIndex(context.Background(), db); err != nil { + _ = base.Close() + return nil, err + } + aligned, err := searchIndexRowsAligned(context.Background(), db) + if err != nil { + _ = base.Close() + return nil, err + } + if !aligned { + if err := rebuildSearchIndexes(context.Background(), db); err != nil { + _ = base.Close() + return nil, fmt.Errorf("align search index rows: %w", err) + } + } + return &Store{db: db, q: storedb.New(db), ftsRowIDAligned: true}, nil } func OpenReadOnly(path string) (*Store, error) { @@ -649,7 +728,24 @@ func OpenReadOnly(path string) (*Store, error) { _ = base.Close() return nil, fmt.Errorf("database schema version %d is newer than this slacrawl build supports (%d)", currentVersion, schemaVersion) } - return &Store{db: db, q: storedb.New(db)}, nil + aligned := false + if currentVersion >= 6 { + pending, err := searchIndexRepairPending(context.Background(), db) + if err != nil { + _ = base.Close() + return nil, err + } + if pending { + _ = base.Close() + return nil, errors.New("database search index rebuild is pending; open it writable to repair") + } + aligned, err = searchIndexRowsAligned(context.Background(), db) + if err != nil { + _ = base.Close() + return nil, err + } + } + return &Store{db: db, q: storedb.New(db), ftsRowIDAligned: aligned}, nil } func (s *Store) DB() *sql.DB { @@ -680,7 +776,11 @@ func (s *Store) UpsertWorkspace(ctx context.Context, workspace Workspace) error // EnsureWorkspace inserts sparse provider metadata without replacing richer data // already collected by another source. func (s *Store) EnsureWorkspace(ctx context.Context, workspace Workspace) error { - _, err := s.db.ExecContext(ctx, ` + return ensureWorkspace(ctx, s.db, workspace) +} + +func ensureWorkspace(ctx context.Context, dbtx storedb.DBTX, workspace Workspace) error { + _, err := dbtx.ExecContext(ctx, ` insert into workspaces (id, name, domain, enterprise_id, raw_json, updated_at) values (?, ?, ?, ?, ?, ?) on conflict(id) do nothing @@ -737,7 +837,11 @@ func (s *Store) UpsertChannel(ctx context.Context, channel Channel) error { // EnsureChannel inserts lower-fidelity provider metadata without replacing an // existing channel collected by a richer source. func (s *Store) EnsureChannel(ctx context.Context, channel Channel) error { - result, err := s.db.ExecContext(ctx, ` + return ensureChannel(ctx, s.db, channel) +} + +func ensureChannel(ctx context.Context, dbtx storedb.DBTX, channel Channel) error { + result, err := dbtx.ExecContext(ctx, ` insert into channels (id, workspace_id, name, kind, topic, purpose, is_private, is_archived, is_shared, is_general, raw_json, updated_at) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) on conflict(id) do nothing @@ -749,7 +853,7 @@ on conflict(id) do nothing if err != nil || rows > 0 { return err } - existing, err := s.getChannelWorkspaceKind(ctx, channel.ID) + existing, err := getChannelWorkspaceKind(ctx, dbtx, channel.ID) if err != nil { return err } @@ -765,11 +869,19 @@ type channelWorkspaceKind struct { } func (s *Store) getChannelWorkspaceKind(ctx context.Context, channelID string) (channelWorkspaceKind, error) { + return getChannelWorkspaceKind(ctx, s.db, channelID) +} + +func getChannelWorkspaceKind(ctx context.Context, q queryRower, channelID string) (channelWorkspaceKind, error) { var row channelWorkspaceKind - err := s.db.QueryRowContext(ctx, `select workspace_id, kind from channels where id = ?`, channelID).Scan(&row.WorkspaceID, &row.Kind) + err := q.QueryRowContext(ctx, `select workspace_id, kind from channels where id = ?`, channelID).Scan(&row.WorkspaceID, &row.Kind) return row, err } +func (s *Store) ChannelWorkspaceID(ctx context.Context, channelID string) (string, error) { + return s.q.GetChannelWorkspace(ctx, channelID) +} + func (s *Store) replaceChannel(ctx context.Context, channel Channel) error { _, err := s.db.ExecContext(ctx, ` update channels @@ -813,13 +925,32 @@ func (s *Store) UpsertUser(ctx context.Context, user User) error { return nil } +func (s *Store) UserWorkspaceID(ctx context.Context, userID string) (string, error) { + return s.q.GetUserWorkspace(ctx, userID) +} + // EnsureUser inserts lower-fidelity provider metadata without replacing an // existing user collected by a richer source. func (s *Store) EnsureUser(ctx context.Context, user User) error { - result, err := s.db.ExecContext(ctx, ` + return ensureUser(ctx, s.db, user) +} + +func ensureUser(ctx context.Context, dbtx storedb.DBTX, user User) error { + result, err := dbtx.ExecContext(ctx, ` insert into users (id, workspace_id, name, real_name, display_name, title, is_bot, is_deleted, raw_json, updated_at) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) -on conflict(id) do nothing +on conflict(id) do update set + workspace_id = excluded.workspace_id, + name = excluded.name, + real_name = excluded.real_name, + display_name = excluded.display_name, + title = excluded.title, + is_bot = excluded.is_bot, + is_deleted = excluded.is_deleted, + raw_json = excluded.raw_json, + updated_at = excluded.updated_at +where users.workspace_id = excluded.workspace_id + and users.raw_json = '`+PlaceholderUserRawJSON+`' `, user.ID, user.WorkspaceID, user.Name, dbText(user.RealName), dbText(user.DisplayName), dbText(user.Title), boolInt(user.IsBot), boolInt(user.IsDeleted), user.RawJSON, formatDBTime(user.UpdatedAt)) if err != nil { return err @@ -828,7 +959,7 @@ on conflict(id) do nothing if err != nil || rows > 0 { return err } - existingWorkspaceID, err := s.q.GetUserWorkspace(ctx, user.ID) + existingWorkspaceID, err := storedb.New(dbtx).GetUserWorkspace(ctx, user.ID) if err != nil { return err } @@ -838,6 +969,309 @@ on conflict(id) do nothing return nil } +// ApplyWriteBatch commits normalized archive records and their sync cursor as +// one unit. Message writes share the single-message priority, retention, FTS, +// file, mention, and event path. +func (s *Store) ApplyWriteBatch(ctx context.Context, batch WriteBatch) (WriteBatchResult, error) { + // BEGIN IMMEDIATE keeps the batched exact-state preflight and subsequent + // sequential writes in one serialized writer snapshot. + dbtx, commit, rollback, err := s.beginMessageTransaction(ctx, true) + if err != nil { + return WriteBatchResult{}, err + } + defer rollback() + qtx := storedb.New(dbtx) + for _, workspace := range batch.Workspaces { + if err := ensureWorkspace(ctx, dbtx, workspace); err != nil { + return WriteBatchResult{}, err + } + } + for _, channel := range batch.Channels { + if err := ensureChannel(ctx, dbtx, channel); err != nil { + return WriteBatchResult{}, err + } + } + for _, user := range batch.Users { + if err := ensureUser(ctx, dbtx, user); err != nil { + return WriteBatchResult{}, err + } + } + unchangedMessages, err := unchangedProviderMessages(ctx, dbtx, batch.Messages) + if err != nil { + return WriteBatchResult{}, err + } + result := WriteBatchResult{} + for _, write := range batch.Messages { + if _, unchanged := unchangedMessages[providerMessageKey(write.Message)]; unchanged { + continue + } + written, err := upsertMessageInTransaction(ctx, dbtx, qtx, write.Message, write.Mentions, write.PreserveHigherPriority, write.EnforceRetention) + if err != nil { + return WriteBatchResult{}, err + } + if written { + result.MessagesWritten++ + } + } + for _, state := range batch.SyncStates { + if err := qtx.SetSyncState(ctx, storedb.SetSyncStateParams{ + SourceName: state.SourceName, EntityType: state.EntityType, EntityID: state.EntityID, + Value: state.Value, UpdatedAt: formatDBTime(time.Now().UTC()), + }); err != nil { + return WriteBatchResult{}, err + } + } + if err := commit(); err != nil { + return WriteBatchResult{}, err + } + return result, nil +} + +const providerPreflightChunkSize = 499 + +type providerMessageIdentity struct { + channelID string + ts string +} + +type providerMessageState struct { + workspaceID string + userID string + subtype string + clientMsgID string + threadTS string + parentUserID string + text string + normalizedText string + replyCount int64 + latestReply string + editedTS string + deletedTS string + sourceRank int64 + sourceName string + rawJSON string +} + +type providerMentionIdentity struct { + mentionType string + targetID string +} + +type providerEventHead struct { + eventType string + sourceName string + payloadJSON string +} + +type providerMessageCandidate struct { + key providerMessageIdentity + message Message + mentions []Mention +} + +// unchangedProviderMessages batches provider replay checks. Duplicate message +// keys deliberately stay on the sequential write path because an earlier +// write in the same batch may change the state seen by a later write. +func unchangedProviderMessages(ctx context.Context, dbtx storedb.DBTX, writes []MessageWrite) (map[providerMessageIdentity]struct{}, error) { + keyCounts := make(map[providerMessageIdentity]int, len(writes)) + for _, write := range writes { + keyCounts[providerMessageKey(write.Message)]++ + } + + candidates := make([]providerMessageCandidate, 0, len(writes)) + for _, write := range writes { + key := providerMessageKey(write.Message) + if write.SkipUnchangedProviderRow && write.Message.Files == nil && keyCounts[key] == 1 { + candidates = append(candidates, providerMessageCandidate{key: key, message: write.Message, mentions: write.Mentions}) + } + } + + unchanged := make(map[providerMessageIdentity]struct{}, len(candidates)) + for start := 0; start < len(candidates); start += providerPreflightChunkSize { + end := min(start+providerPreflightChunkSize, len(candidates)) + chunk := candidates[start:end] + keys := make([]providerMessageIdentity, len(chunk)) + for i := range chunk { + keys[i] = chunk[i].key + } + + states, err := loadProviderMessageStates(ctx, dbtx, keys) + if err != nil { + return nil, err + } + mentions, err := loadProviderMessageMentions(ctx, dbtx, keys) + if err != nil { + return nil, err + } + heads, err := loadProviderMessageEventHeads(ctx, dbtx, keys) + if err != nil { + return nil, err + } + + for _, candidate := range chunk { + state, exists := states[candidate.key] + if !exists || state != providerMessageStateFromMessage(candidate.message) { + continue + } + if !providerMentionsEqual(mentions[candidate.key], candidate.mentions) { + continue + } + expectedHead := providerEventHead{ + eventType: eventType(candidate.message), + sourceName: candidate.message.SourceName, + payloadJSON: candidate.message.RawJSON, + } + if _, ok := heads[candidate.key][expectedHead]; !ok { + continue + } + unchanged[candidate.key] = struct{}{} + } + } + return unchanged, nil +} + +func providerMessageKey(message Message) providerMessageIdentity { + return providerMessageIdentity{channelID: message.ChannelID, ts: message.TS} +} + +func providerMessageStateFromMessage(message Message) providerMessageState { + return providerMessageState{ + workspaceID: message.WorkspaceID, userID: message.UserID, subtype: message.Subtype, + clientMsgID: message.ClientMsgID, threadTS: message.ThreadTS, parentUserID: message.ParentUserID, + text: message.Text, normalizedText: message.NormalizedText, replyCount: int64(message.ReplyCount), + latestReply: message.LatestReply, editedTS: message.EditedTS, deletedTS: message.DeletedTS, + sourceRank: int64(message.SourceRank), sourceName: message.SourceName, rawJSON: message.RawJSON, + } +} + +func loadProviderMessageStates(ctx context.Context, dbtx storedb.DBTX, keys []providerMessageIdentity) (map[providerMessageIdentity]providerMessageState, error) { + predicate, args := providerMessageKeyPredicate(keys) + rows, err := dbtx.QueryContext(ctx, ` +select channel_id, ts, workspace_id, coalesce(user_id, ''), coalesce(subtype, ''), + coalesce(client_msg_id, ''), coalesce(thread_ts, ''), coalesce(parent_user_id, ''), + text, normalized_text, reply_count, coalesce(latest_reply, ''), coalesce(edited_ts, ''), + coalesce(deleted_ts, ''), source_rank, source_name, raw_json +from messages where `+predicate, args...) + if err != nil { + return nil, fmt.Errorf("read provider message states: %w", err) + } + states := make(map[providerMessageIdentity]providerMessageState, len(keys)) + for rows.Next() { + var key providerMessageIdentity + var state providerMessageState + if err := rows.Scan( + &key.channelID, &key.ts, &state.workspaceID, &state.userID, &state.subtype, + &state.clientMsgID, &state.threadTS, &state.parentUserID, &state.text, + &state.normalizedText, &state.replyCount, &state.latestReply, &state.editedTS, + &state.deletedTS, &state.sourceRank, &state.sourceName, &state.rawJSON, + ); err != nil { + _ = rows.Close() + return nil, fmt.Errorf("scan provider message state: %w", err) + } + states[key] = state + } + if err := rows.Err(); err != nil { + _ = rows.Close() + return nil, fmt.Errorf("read provider message states: %w", err) + } + if err := rows.Close(); err != nil { + return nil, fmt.Errorf("close provider message states: %w", err) + } + return states, nil +} + +func loadProviderMessageMentions(ctx context.Context, dbtx storedb.DBTX, keys []providerMessageIdentity) (map[providerMessageIdentity]map[providerMentionIdentity]string, error) { + predicate, args := providerMessageKeyPredicate(keys) + rows, err := dbtx.QueryContext(ctx, ` +select channel_id, ts, mention_type, target_id, coalesce(display_text, '') +from message_mentions where `+predicate, args...) + if err != nil { + return nil, fmt.Errorf("read provider message mentions: %w", err) + } + mentions := make(map[providerMessageIdentity]map[providerMentionIdentity]string) + for rows.Next() { + var key providerMessageIdentity + var mention providerMentionIdentity + var displayText string + if err := rows.Scan(&key.channelID, &key.ts, &mention.mentionType, &mention.targetID, &displayText); err != nil { + _ = rows.Close() + return nil, fmt.Errorf("scan provider message mention: %w", err) + } + if mentions[key] == nil { + mentions[key] = make(map[providerMentionIdentity]string) + } + mentions[key][mention] = displayText + } + if err := rows.Err(); err != nil { + _ = rows.Close() + return nil, fmt.Errorf("read provider message mentions: %w", err) + } + if err := rows.Close(); err != nil { + return nil, fmt.Errorf("close provider message mentions: %w", err) + } + return mentions, nil +} + +func loadProviderMessageEventHeads(ctx context.Context, dbtx storedb.DBTX, keys []providerMessageIdentity) (map[providerMessageIdentity]map[providerEventHead]struct{}, error) { + predicate, args := providerMessageKeyPredicate(keys) + rows, err := dbtx.QueryContext(ctx, ` +select channel_id, ts, event_type, source_name, payload_json +from message_event_heads where `+predicate, args...) + if err != nil { + return nil, fmt.Errorf("read provider message event heads: %w", err) + } + heads := make(map[providerMessageIdentity]map[providerEventHead]struct{}) + for rows.Next() { + var key providerMessageIdentity + var head providerEventHead + if err := rows.Scan(&key.channelID, &key.ts, &head.eventType, &head.sourceName, &head.payloadJSON); err != nil { + _ = rows.Close() + return nil, fmt.Errorf("scan provider message event head: %w", err) + } + if heads[key] == nil { + heads[key] = make(map[providerEventHead]struct{}) + } + heads[key][head] = struct{}{} + } + if err := rows.Err(); err != nil { + _ = rows.Close() + return nil, fmt.Errorf("read provider message event heads: %w", err) + } + if err := rows.Close(); err != nil { + return nil, fmt.Errorf("close provider message event heads: %w", err) + } + return heads, nil +} + +func providerMentionsEqual(actual map[providerMentionIdentity]string, expected []Mention) bool { + expectedByID := make(map[providerMentionIdentity]string, len(expected)) + for _, mention := range expected { + expectedByID[providerMentionIdentity{mentionType: mention.Type, targetID: mention.TargetID}] = mention.DisplayText + } + if len(actual) != len(expectedByID) { + return false + } + for key, displayText := range expectedByID { + if actualDisplayText, ok := actual[key]; !ok || actualDisplayText != displayText { + return false + } + } + return true +} + +func providerMessageKeyPredicate(keys []providerMessageIdentity) (string, []any) { + var predicate strings.Builder + args := make([]any, 0, len(keys)*2) + for i, key := range keys { + if i > 0 { + predicate.WriteString(" or ") + } + predicate.WriteString("(channel_id = ? and ts = ?)") + args = append(args, key.channelID, key.ts) + } + return predicate.String(), args +} + func (s *Store) UpsertMessage(ctx context.Context, message Message, mentions []Mention) error { _, err := s.upsertMessage(ctx, message, mentions, false, false) return err @@ -857,12 +1291,26 @@ func (s *Store) UpsertMessageByPriorityWithRetention(ctx context.Context, messag } func (s *Store) upsertMessage(ctx context.Context, message Message, mentions []Mention, preserveHigherPriority, enforceRetention bool) (bool, error) { - key := messageKey(message.ChannelID, message.TS) dbtx, commit, rollback, err := s.beginMessageTransaction(ctx, enforceRetention) if err != nil { return false, err } defer rollback() + written, err := upsertMessageInTransaction(ctx, dbtx, storedb.New(dbtx), message, mentions, preserveHigherPriority, enforceRetention) + if err != nil { + return false, err + } + if !written { + return false, nil + } + if err := commit(); err != nil { + return false, err + } + return written, nil +} + +func upsertMessageInTransaction(ctx context.Context, dbtx storedb.DBTX, qtx *storedb.Queries, message Message, mentions []Mention, preserveHigherPriority, enforceRetention bool) (bool, error) { + key := messageKey(message.ChannelID, message.TS) if enforceRetention { allowed, err := messageAllowedByRetention(ctx, dbtx, message) if err != nil { @@ -872,8 +1320,7 @@ func (s *Store) upsertMessage(ctx context.Context, message Message, mentions []M return false, nil } } - qtx := storedb.New(dbtx) - + var err error var rows int64 if preserveHigherPriority { rows, err = qtx.UpsertMessageByPriority(ctx, storedb.UpsertMessageByPriorityParams{ @@ -957,12 +1404,9 @@ func (s *Store) upsertMessage(ctx context.Context, message Message, mentions []M } } - if err := qtx.DeleteMessageFTS(ctx, key); err != nil { - return false, err - } searchMessage := message searchMessage.Files = filesForSearch - if err := qtx.InsertMessageFTS(ctx, storedb.InsertMessageFTSParams{MessageKey: key, Content: messageSearchContent(searchMessage)}); err != nil { + if err := writeMessageFTS(ctx, dbtx, message.ChannelID, message.TS, messageSearchContent(searchMessage)); err != nil { return false, err } @@ -970,12 +1414,28 @@ func (s *Store) upsertMessage(ctx context.Context, message Message, mentions []M return false, err } - if err := commit(); err != nil { - return false, err - } return true, nil } +type messageFTSWriter interface { + ExecContext(context.Context, string, ...any) (sql.Result, error) +} + +func writeMessageFTS(ctx context.Context, writer messageFTSWriter, channelID, ts, content string) error { + result, err := writer.ExecContext(ctx, upsertMessageFTSRowSQL, content, channelID, ts) + if err != nil { + return err + } + rows, err := result.RowsAffected() + if err != nil { + return err + } + if rows != 1 { + return fmt.Errorf("message %q search index upsert affected %d rows", messageKey(channelID, ts), rows) + } + return nil +} + func (s *Store) MarkMessageDeleted(ctx context.Context, message Message, mentions []Mention) error { _, err := s.markMessageDeleted(ctx, message, mentions, false) return err @@ -1019,7 +1479,7 @@ select exists ( return false, err } } - if _, err := dbtx.ExecContext(ctx, `delete from message_fts where message_key = ?`, messageKey(channelID, ts)); err != nil { + if _, err := dbtx.ExecContext(ctx, deleteMessageFTSRowSQL, channelID, ts); err != nil { return false, err } if _, err := dbtx.ExecContext(ctx, ` @@ -1101,12 +1561,9 @@ func (s *Store) markMessageDeleted(ctx context.Context, message Message, mention } return false, fmt.Errorf("message %q upsert affected no rows", key) } - if err := qtx.DeleteMessageFTS(ctx, key); err != nil { - return false, err - } searchMessage := message searchMessage.Files = nil - if err := qtx.InsertMessageFTS(ctx, storedb.InsertMessageFTSParams{MessageKey: key, Content: messageSearchContent(searchMessage)}); err != nil { + if err := writeMessageFTS(ctx, dbtx, message.ChannelID, message.TS, messageSearchContent(searchMessage)); err != nil { return false, err } if err := replaceMessageMentions(ctx, qtx, message, mentions); err != nil { @@ -1120,10 +1577,7 @@ func (s *Store) markMessageDeleted(ctx context.Context, message Message, mention searchMessage := message searchMessage.NormalizedText = normalizedText searchMessage.Files = nil - if err := qtx.DeleteMessageFTS(ctx, key); err != nil { - return false, err - } - if err := qtx.InsertMessageFTS(ctx, storedb.InsertMessageFTSParams{MessageKey: key, Content: messageSearchContent(searchMessage)}); err != nil { + if err := writeMessageFTS(ctx, dbtx, message.ChannelID, message.TS, messageSearchContent(searchMessage)); err != nil { return false, err } } @@ -1520,17 +1974,31 @@ func (s *Store) searchAuto(ctx context.Context, workspaceID string, query string } func (s *Store) searchFTS(ctx context.Context, workspaceID string, query string, limit int) ([]MessageRow, error) { + if s.searchIndexUnavailable.Load() { + return nil, errors.New("search index is temporarily unavailable; retry or reopen the database") + } + messageJoin := "join messages m on f.message_key = m.channel_id || '|' || m.ts" + if s.ftsRowIDAligned { + messageJoin = "join messages m on f.rowid = m.rowid" + } sqlQuery := ` select ` + messageRowSelect + ` from message_fts f -join messages m on f.message_key = m.channel_id || '|' || m.ts +` + messageJoin + ` ` + messageRowJoins + ` where message_fts match ? and (? = '' or m.workspace_id = ?) order by m.ts desc limit ? ` - return s.queryMessageRows(ctx, sqlQuery, query, workspaceID, workspaceID, RequireLimit(limit)) + rows, err := s.queryMessageRows(ctx, sqlQuery, query, workspaceID, workspaceID, RequireLimit(limit)) + if err != nil { + return nil, err + } + if s.searchIndexUnavailable.Load() { + return nil, errors.New("search index is temporarily unavailable; retry or reopen the database") + } + return rows, nil } func (s *Store) searchLike(ctx context.Context, workspaceID string, query string, limit int) ([]MessageRow, error) { @@ -1717,12 +2185,12 @@ func (s *Store) resolveMessageRowMentions(ctx context.Context, rows []MessageRow if len(rows) == 0 { return nil } - byKey := map[string][]messageMentionDisplay{} - keys := make([]string, 0, len(rows)) - seen := map[string]struct{}{} + byKey := map[providerMessageIdentity][]messageMentionDisplay{} + keys := make([]providerMessageIdentity, 0, len(rows)) + seen := map[providerMessageIdentity]struct{}{} for _, row := range rows { - key := messageKey(row.ChannelID, row.TS) - if strings.TrimSpace(key) == "|" { + key := providerMessageIdentity{channelID: row.ChannelID, ts: row.TS} + if strings.TrimSpace(key.channelID) == "" && strings.TrimSpace(key.ts) == "" { continue } if _, ok := seen[key]; ok { @@ -1733,7 +2201,7 @@ func (s *Store) resolveMessageRowMentions(ctx context.Context, rows []MessageRow } for start := 0; start < len(keys); start += 400 { end := min(start+400, len(keys)) - placeholders := strings.TrimRight(strings.Repeat("?,", end-start), ",") + predicate, args := providerMessageKeyPredicate(keys[start:end]) query := ` select mm.channel_id, mm.ts, mm.target_id, coalesce(nullif(u.display_name, ''), nullif(u.real_name, ''), nullif(u.name, ''), nullif(mm.display_text, ''), '') @@ -1741,12 +2209,8 @@ from message_mentions mm left join users u on u.id = mm.target_id where mm.mention_type = 'user' and mm.deleted_at is null - and (mm.channel_id || '|' || mm.ts) in (` + placeholders + `) + and (` + predicate + `) ` - args := make([]any, 0, end-start) - for _, key := range keys[start:end] { - args = append(args, key) - } mentionRows, err := s.db.QueryContext(ctx, query, args...) if err != nil { return err @@ -1762,7 +2226,7 @@ where mm.mention_type = 'user' if target == "" || display == "" || strings.EqualFold(display, target) { continue } - key := messageKey(channelID, ts) + key := providerMessageIdentity{channelID: channelID, ts: ts} byKey[key] = append(byKey[key], messageMentionDisplay{target: target, display: display}) } if err := mentionRows.Err(); err != nil { @@ -1774,7 +2238,7 @@ where mm.mention_type = 'user' } } for index := range rows { - key := messageKey(rows[index].ChannelID, rows[index].TS) + key := providerMessageIdentity{channelID: rows[index].ChannelID, ts: rows[index].TS} mentions := byKey[key] if len(mentions) == 0 { continue @@ -2350,6 +2814,99 @@ func (s *Store) ListSyncState(ctx context.Context, source, entityType string, li return out, nil } +func (s *Store) RebuildSearchIndexes(ctx context.Context) error { + return rebuildSearchIndexes(ctx, s.db) +} + +// RebuildSearchIndexesInTransaction atomically aligns FTS rows with messages +// while preserving the caller's surrounding database transaction. +func RebuildSearchIndexesInTransaction(ctx context.Context, tx *sql.Tx) error { + if tx == nil { + return errors.New("search-index rebuild transaction is required") + } + return rebuildSearchIndexesInTransaction(ctx, tx) +} + +func markSearchIndexRebuildPending(ctx context.Context, dbtx storedb.DBTX) error { + return storedb.New(dbtx).SetSyncState(ctx, storedb.SetSyncStateParams{ + SourceName: searchIndexMaintenanceSource, EntityType: searchIndexMaintenanceEntityType, + EntityID: searchIndexMaintenanceEntityID, Value: "1", UpdatedAt: formatDBTime(time.Now().UTC()), + }) +} + +func repairPendingSearchIndex(ctx context.Context, db *sql.DB) error { + pending, err := searchIndexRepairPending(ctx, db) + if err != nil { + return err + } + if !pending { + return nil + } + if err := rebuildSearchIndexes(ctx, db); err != nil { + return fmt.Errorf("repair pending search index: %w", err) + } + return nil +} + +func searchIndexRepairPending(ctx context.Context, db *sql.DB) (bool, error) { + var pending bool + if err := db.QueryRowContext(ctx, ` +select exists ( + select 1 from sync_state + where source_name = ? and entity_type = ? and entity_id = ? +) +`, searchIndexMaintenanceSource, searchIndexMaintenanceEntityType, searchIndexMaintenanceEntityID).Scan(&pending); err != nil { + return false, fmt.Errorf("inspect search-index maintenance state: %w", err) + } + return pending, nil +} + +func searchIndexRowsAligned(ctx context.Context, db *sql.DB) (bool, error) { + var aligned bool + if err := db.QueryRowContext(ctx, ` +select + (select count(*) from message_fts) = (select count(*) from messages) + and not exists ( + select 1 + from message_fts f + left join messages m + on m.rowid = f.rowid + and f.message_key = m.channel_id || '|' || m.ts + where m.rowid is null + ) +`).Scan(&aligned); err != nil { + return false, fmt.Errorf("inspect search-index row alignment: %w", err) + } + return aligned, nil +} + +func rebuildSearchIndexes(ctx context.Context, db *sql.DB) error { + tx, err := db.BeginTx(ctx, nil) + if err != nil { + return err + } + defer func() { _ = tx.Rollback() }() + if err := rebuildSearchIndexesInTransaction(ctx, tx); err != nil { + return err + } + return tx.Commit() +} + +func rebuildSearchIndexesInTransaction(ctx context.Context, dbtx storedb.DBTX) error { + if _, err := dbtx.ExecContext(ctx, `delete from message_fts`); err != nil { + return err + } + if _, err := dbtx.ExecContext(ctx, rebuildMessageFTSRowsSQL); err != nil { + return err + } + if _, err := dbtx.ExecContext(ctx, ` +delete from sync_state +where source_name = ? and entity_type = ? and entity_id = ? +`, searchIndexMaintenanceSource, searchIndexMaintenanceEntityType, searchIndexMaintenanceEntityID); err != nil { + return err + } + return nil +} func MarshalRaw(v any) string { data, err := json.Marshal(v) if err != nil { @@ -2704,19 +3261,18 @@ where exists ( and trim(coalesce(m.deleted_ts, '')) <> '' ); -delete from message_fts; -insert into message_fts (message_key, content) -select m.channel_id || '|' || m.ts, - trim(m.normalized_text || ' ' || coalesce(( - select group_concat(trim(f.name || ' ' || f.title || ' ' || f.plain_text || ' ' || f.preview_plain_text), ' ') - from message_files f - where f.channel_id = m.channel_id and f.ts = m.ts and f.deleted_at is null - ), '')) -from messages m; `); err != nil { return err } - _, err = tx.Exec(schemaV6Migration) + if _, err := tx.Exec(`delete from message_fts;` + rebuildMessageFTSRowsSQL + `; +delete from sync_state +where source_name = '` + searchIndexMaintenanceSource + `' + and entity_type = '` + searchIndexMaintenanceEntityType + `' + and entity_id = '` + searchIndexMaintenanceEntityID + `'; +`); err != nil { + return err + } + _, err = tx.Exec(schemaV6EventMigration) return err } diff --git a/internal/store/store_test.go b/internal/store/store_test.go index df58423..987050a 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -1035,6 +1035,92 @@ pragma user_version = 5; require.Contains(t, defaultValue, "randomblob") } +func TestOpenMigratesVersion5FTSRowIDs(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "test.db") + s, err := Open(dbPath) + require.NoError(t, err) + ctx := context.Background() + now := time.Now().UTC() + seedBatchCatalog(t, s, "T1", "C1", "U1", now) + require.NoError(t, s.UpsertMessage(ctx, Message{ + ChannelID: "C1", TS: "1.000001", WorkspaceID: "T1", UserID: "U1", + Text: "first", NormalizedText: "first", SourceRank: 2, SourceName: "api-bot", + RawJSON: "{}", UpdatedAt: now, + Files: []MessageFile{{FileID: "F1", Name: "notes.txt", PlainText: "migrated appendix", RawJSON: "{}"}}, + }, nil)) + require.NoError(t, s.UpsertMessage(ctx, batchMessage("C1", "2.000002", "T1", "second", now), nil)) + require.NoError(t, s.Close()) + + db, err := sql.Open("sqlite", dbPath) + require.NoError(t, err) + _, err = db.Exec(` +delete from message_fts; +insert into message_fts (rowid, message_key, content) values + (1001, 'C1|1.000001', 'legacy poison one'), + (1002, 'C1|2.000002', 'legacy poison two'), + (1003, 'C1|orphan', 'legacy poison orphan'); +pragma user_version = 5; +`) + require.NoError(t, err) + require.NoError(t, db.Close()) + + s, err = Open(dbPath) + require.NoError(t, err) + defer func() { require.NoError(t, s.Close()) }() + var version int + require.NoError(t, s.DB().QueryRow(`pragma user_version`).Scan(&version)) + require.Equal(t, schemaVersion, version) + requireMessageFTSParity(t, s) + matches, err := s.Search(ctx, "T1", "appendix", 10) + require.NoError(t, err) + require.Len(t, matches, 1) + assertBatchCount(t, s, `select count(*) from message_fts where content like '%poison%'`, 0) +} + +func TestMessageFTSRowIDParityAcrossMutationAndRebuild(t *testing.T) { + s := openBatchTestStore(t) + ctx := context.Background() + now := time.Now().UTC() + seedBatchCatalog(t, s, "T1", "C1", "U1", now) + first := batchMessage("C1", "1.000001", "T1", "first", now) + second := batchMessage("C1", "2.000002", "T1", "second", now) + third := batchMessage("C1", "3.000003", "T1", "third", now) + third.Files = []MessageFile{{FileID: "F3", Name: "third.txt", PlainText: "rebuild appendix", RawJSON: "{}"}} + for _, message := range []Message{first, second, third} { + require.NoError(t, s.UpsertMessage(ctx, message, nil)) + } + requireMessageFTSParity(t, s) + + deleted := first + deleted.DeletedTS = "4.000004" + deleted.UpdatedAt = now.Add(time.Second) + require.NoError(t, s.MarkMessageDeleted(ctx, deleted, nil)) + requireMessageFTSParity(t, s) + + removed, err := s.DeleteMessageBySource(ctx, "T1", "C1", second.TS, second.SourceName) + require.NoError(t, err) + require.True(t, removed) + removed, err = s.DeleteMessageBySource(ctx, "T1", "C1", second.TS, second.SourceName) + require.NoError(t, err) + require.False(t, removed) + requireMessageFTSParity(t, s) + + _, err = s.DB().ExecContext(ctx, ` +delete from message_fts; +insert into message_fts (rowid, message_key, content) values + (1001, 'C1|1.000001', 'wrong first'), + (1003, 'C1|3.000003', 'wrong third'), + (1004, 'C1|orphan', 'wrong orphan'); +`) + require.NoError(t, err) + require.NoError(t, s.RebuildSearchIndexes(ctx)) + requireMessageFTSParity(t, s) + matches, err := s.Search(ctx, "T1", "appendix", 10) + require.NoError(t, err) + require.Len(t, matches, 1) + require.Equal(t, third.TS, matches[0].TS) +} + func TestOpenDoesNotStampInvalidOldSchema(t *testing.T) { dbPath := filepath.Join(t.TempDir(), "test.db") db, err := sql.Open("sqlite", dbPath) diff --git a/internal/store/storedb/queries.sql.go b/internal/store/storedb/queries.sql.go index 2856872..8cb11ef 100644 --- a/internal/store/storedb/queries.sql.go +++ b/internal/store/storedb/queries.sql.go @@ -135,15 +135,6 @@ func (q *Queries) CountWorkspaces(ctx context.Context) (int64, error) { return count, err } -const deleteMessageFTS = `-- name: DeleteMessageFTS :exec -delete from message_fts where message_key = ? -` - -func (q *Queries) DeleteMessageFTS(ctx context.Context, messageKey string) error { - _, err := q.db.ExecContext(ctx, deleteMessageFTS, messageKey) - return err -} - const deleteMessageFiles = `-- name: DeleteMessageFiles :exec delete from message_files where channel_id = ? and ts = ? ` @@ -332,20 +323,6 @@ func (q *Queries) InsertMessageEvent(ctx context.Context, arg InsertMessageEvent return err } -const insertMessageFTS = `-- name: InsertMessageFTS :exec -insert into message_fts (message_key, content) values (?, ?) -` - -type InsertMessageFTSParams struct { - MessageKey string `json:"message_key"` - Content string `json:"content"` -} - -func (q *Queries) InsertMessageFTS(ctx context.Context, arg InsertMessageFTSParams) error { - _, err := q.db.ExecContext(ctx, insertMessageFTS, arg.MessageKey, arg.Content) - return err -} - const insertMessageFile = `-- name: InsertMessageFile :exec insert into message_files ( workspace_id, channel_id, ts, file_id, user_id, name, title, mimetype, filetype, diff --git a/internal/syncer/syncer.go b/internal/syncer/syncer.go index 507b847..5311856 100644 --- a/internal/syncer/syncer.go +++ b/internal/syncer/syncer.go @@ -9,6 +9,7 @@ import ( "strings" "github.com/openclaw/slacrawl/internal/config" + "github.com/openclaw/slacrawl/internal/provider" "github.com/openclaw/slacrawl/internal/slackapi" "github.com/openclaw/slacrawl/internal/slackdesktop" "github.com/openclaw/slacrawl/internal/slackmcp" @@ -25,7 +26,8 @@ const ( ) func ParseSource(value string) (Source, error) { - switch strings.ToLower(strings.TrimSpace(value)) { + normalized := strings.ToLower(strings.TrimSpace(value)) + switch normalized { case "", string(SourceAPI), "bot": return SourceAPI, nil case string(SourceDesktop), "wiretap": @@ -34,9 +36,16 @@ func ParseSource(value string) (Source, error) { return SourceMCP, nil case string(SourceAll), "hybrid": return SourceAll, nil - default: - return "", fmt.Errorf("unsupported source %q: use api, bot, desktop, wiretap, mcp, connector, or all", value) } + if name, ok := strings.CutPrefix(normalized, "provider:"); ok && name != "" && !strings.ContainsAny(name, ":/\\\t\r\n ") { + return Source("provider:" + name), nil + } + return "", fmt.Errorf("unsupported source %q: use api, bot, desktop, wiretap, mcp, connector, all, or provider:", value) +} + +func ProviderName(source Source) (string, bool) { + name, ok := strings.CutPrefix(string(source), "provider:") + return name, ok && name != "" } type Options struct { @@ -48,6 +57,7 @@ type Options struct { Since string Full bool LatestOnly bool + Limit int Concurrency int AutoJoin *bool APIURL string @@ -56,8 +66,9 @@ type Options struct { } type Summary struct { - Desktop slackdesktop.Source `json:"desktop"` - MCP *slackmcp.Summary `json:"mcp,omitempty"` + Desktop slackdesktop.Source `json:"desktop"` + MCP *slackmcp.Summary `json:"mcp,omitempty"` + Provider *provider.Summary `json:"provider,omitempty"` } func Run(ctx context.Context, cfg config.Config, st *store.Store, opts Options) (Summary, error) { @@ -114,7 +125,22 @@ func RunWithTokens(ctx context.Context, cfg config.Config, st *store.Store, opts } return syncDesktop(ctx, cfg, st, desktopOptionsForSourceAll(opts)) default: - return summary, errors.New("unsupported source") + name, ok := ProviderName(opts.Source) + if !ok { + return summary, errors.New("unsupported source") + } + providerConfig, ok := cfg.Provider(name) + if !ok { + return summary, fmt.Errorf("provider %q is not configured", name) + } + providerSummary, err := provider.Sync(ctx, st, providerConfig, provider.Options{ + WorkspaceID: opts.WorkspaceID, Channels: opts.Channels, + ExcludeChannels: opts.ExcludeChannels, Since: opts.Since, + Full: opts.Full, LatestOnly: opts.LatestOnly, Limit: opts.Limit, + Logger: opts.Logger, + }) + summary.Provider = &providerSummary + return summary, err } } diff --git a/internal/syncer/syncer_test.go b/internal/syncer/syncer_test.go index d74f8d8..57a2112 100644 --- a/internal/syncer/syncer_test.go +++ b/internal/syncer/syncer_test.go @@ -30,6 +30,20 @@ func TestParseSourceRejectsUnknown(t *testing.T) { require.ErrorContains(t, err, "unsupported source") } +func TestParseSourceProvider(t *testing.T) { + source, err := ParseSource(" Provider:Archive ") + require.NoError(t, err) + require.Equal(t, Source("provider:archive"), source) + name, ok := ProviderName(source) + require.True(t, ok) + require.Equal(t, "archive", name) + + for _, input := range []string{"provider:", "provider:bad:name", "provider:bad/name"} { + _, err := ParseSource(input) + require.ErrorContains(t, err, "unsupported source") + } +} + func TestDesktopOptionsForSourceAllClearsInheritedWorkspace(t *testing.T) { opts := desktopOptionsForSourceAll(Options{Source: SourceAll, WorkspaceID: "T111"}) require.Empty(t, opts.WorkspaceID)