From 1f361b511b22b36f74c5fd7c878485da8a5080c8 Mon Sep 17 00:00:00 2001 From: Claude Bot Date: Mon, 20 Jul 2026 04:57:55 +0000 Subject: [PATCH] feat(ingest): shared connector builder + opt-in in-process sync scheduler (#428) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Deduplicate connector construction into internal/connectors.Build, reused by the CLI (buildConnector) and MCP (importFunc) as thin wrappers, and add an opt-in background sync in cmd/http. - internal/connectors.Build(ctx, kb, cfg, source, path, ocr) is the single place connectors are built; OAuth resolution (#246) lives inside, OCR is passed in by the caller so the shell-out stays in the cmd layer. CLI + MCP now delegate to it (net -100 lines, one place to edit a connector). - cmd/http autoSync: with SYNC_INTERVAL set (e.g. 15m), re-ingest every configured REMOTE connector on a timer — the in-process form of `kb sync` (#415) — incrementally, honoring INGEST_PRUNE_DELETED. markdown is skipped (autoImport already sweeps it). Default off (empty SYNC_INTERVAL) so the minimal single-user stack is unchanged. - Config: Ingest.SyncInterval + SyncInterval() + SYNC_INTERVAL env. - Tests: connectors.Build construction/error cases (no DB); SyncInterval parse + env. SYSTEM.md decision log + .env.example documented. Push adapters (fsnotify/webhooks) remain a follow-up. Co-Authored-By: Claude Opus 4.8 (1M context) --- .env.example | 6 + SYSTEM.md | 11 ++ cmd/cli/commands.go | 128 +--------------------- cmd/http/main.go | 66 +++++++++++ cmd/mcp/main.go | 136 ++--------------------- internal/config/config.go | 19 ++++ internal/config/config_test.go | 27 +++++ internal/connectors/connectors.go | 146 +++++++++++++++++++++++++ internal/connectors/connectors_test.go | 64 +++++++++++ 9 files changed, 356 insertions(+), 247 deletions(-) create mode 100644 internal/connectors/connectors.go create mode 100644 internal/connectors/connectors_test.go diff --git a/.env.example b/.env.example index 4a40827..13e51c4 100644 --- a/.env.example +++ b/.env.example @@ -28,6 +28,12 @@ EMBEDDING_MODEL=nomic-embed-text # Auto-import: re-scan ./data/docs on this interval (empty = off). Great for laptops. INGEST_INTERVAL=60s +# Background connector sync (optional, #428): re-ingest the configured REMOTE +# connectors (notion, slack, github, …) on this interval — the in-process form of +# `kb sync`. Empty (default) = off, so the minimal single-user stack is unchanged; +# markdown stays governed by INGEST_INTERVAL above. Honors INGEST_PRUNE_DELETED. +# SYNC_INTERVAL=15m + # Propagate source-side deletions on auto-import (optional, #247). OFF by default: # the retention default (#107) keeps content even after you delete the file. When # on, a doc removed from ./data/docs is removed from memory on the next sweep (only diff --git a/SYSTEM.md b/SYSTEM.md index 0651cd5..b3134e3 100644 --- a/SYSTEM.md +++ b/SYSTEM.md @@ -383,6 +383,17 @@ as the adoption signal. Newest first. +- **2026-07-20** — **Shared connector builder + opt-in in-process sync scheduler (#428, ingestion P2).** + The ~150-line connector-construction switch was duplicated in the CLI (`buildConnector`) and the MCP server + (`importFunc`); it now lives once in **`internal/connectors.Build(ctx, kb, cfg, source, path, ocr)`** — + the CLI and MCP adapters are thin wrappers over it, so a new/changed connector is edited in one place. OAuth + token resolution (#246) lives inside the builder; the OCR fallback is passed in by the caller so the + shell-out stays in the cmd layer (the markdown connector is the only consumer). On top of it, `cmd/http` + gains an **opt-in background sync** (`autoSync`, mirrors `autoImport`): with `SYNC_INTERVAL` set (e.g. + `15m`), the HTTP server re-ingests every configured **remote** connector on a timer — the in-process form + of `kb sync` (#415) — incrementally, honoring `INGEST_PRUNE_DELETED`. markdown is skipped (already swept by + `autoImport`). **Default off** (`SYNC_INTERVAL` empty → scheduler never starts), so the minimal single-user + stack is unchanged. Push adapters (fsnotify for markdown, connector webhooks) remain a follow-up. - **2026-07-20** — **Wire async batch extraction into the ingest loop (#430, completes #420, ingestion P2).** With `EXTRACTION_BATCH` + the Claude extractor (`c.batchExtractor` set), ingest's extractor step now submits the document's hot chunks as one Message Batch (`SubmitExtractionBatch`, custom_id = content hash, diff --git a/cmd/cli/commands.go b/cmd/cli/commands.go index 8d59b39..3571726 100644 --- a/cmd/cli/commands.go +++ b/cmd/cli/commands.go @@ -17,19 +17,10 @@ import ( "github.com/programmism/brainiac/internal/chunk" "github.com/programmism/brainiac/internal/config" + "github.com/programmism/brainiac/internal/connectors" "github.com/programmism/brainiac/internal/core" "github.com/programmism/brainiac/internal/model" "github.com/programmism/brainiac/internal/plugins" - "github.com/programmism/brainiac/internal/plugins/confluence" - "github.com/programmism/brainiac/internal/plugins/gdrive" - "github.com/programmism/brainiac/internal/plugins/github" - "github.com/programmism/brainiac/internal/plugins/gitlab" - "github.com/programmism/brainiac/internal/plugins/gmail" - "github.com/programmism/brainiac/internal/plugins/jira" - "github.com/programmism/brainiac/internal/plugins/linear" - "github.com/programmism/brainiac/internal/plugins/markdown" - "github.com/programmism/brainiac/internal/plugins/notion" - "github.com/programmism/brainiac/internal/plugins/slack" "github.com/programmism/brainiac/internal/store" ) @@ -1133,120 +1124,11 @@ func importCmd() *cobra.Command { return cmd } -// oauthToken resolves a connector's access token (#246): a stored, auto-refreshed -// OAuth credential if present, else the source's _TOKEN env value. -func oauthToken(ctx context.Context, kb *core.Core, cfg *config.Config, source string) (string, error) { - env := "" - if sc := cfg.Source(source); sc != nil { - env = sc.Token - } - return kb.ResolveSourceToken(ctx, source, env) -} - +// buildConnector constructs the connector for a source type via the shared +// internal/connectors builder (#428), passing the CLI's OCR fallback. path is an +// optional targeted import (--path). func buildConnector(ctx context.Context, kb *core.Core, cfg *config.Config, source, path string) (plugins.SourceConnector, error) { - switch source { - case "notion": - sc := cfg.Source("notion") - if sc == nil || sc.Token == "" { - return nil, fmt.Errorf("notion source not configured (set a token via NOTION_TOKEN or config.yaml)") - } - if path != "" { // --path holds a page URL/id for a targeted import - return notion.NewForPages(sc.Token, []string{path}), nil - } - return notion.New(sc.Token), nil - case "slack": - sc := cfg.Source("slack") - if sc == nil || sc.Token == "" { - return nil, fmt.Errorf("slack source not configured (set a token via SLACK_TOKEN or config.yaml)") - } - if path != "" { // --path holds a channel id for a targeted import - return slack.NewForChannels(sc.Token, []string{path}), nil - } - return slack.New(sc.Token), nil - case "github": - sc := cfg.Source("github") - if sc == nil || sc.Token == "" { - return nil, fmt.Errorf("github source not configured (set a token via GITHUB_TOKEN or config.yaml)") - } - repos := sc.Repos - if path != "" { // --path holds an owner/repo for a targeted import - repos = []string{path} - } - if len(repos) == 0 { - return nil, fmt.Errorf("github needs a repo: --path owner/repo, or sources[].repos / GITHUB_REPOS") - } - ghOpts := []github.Option{github.WithFiles(sc.Files)} - if sc.Discussions { - ghOpts = append(ghOpts, github.WithDiscussions()) - } - return github.New(sc.Token, repos, ghOpts...), nil - case "gdrive": - tok, err := oauthToken(ctx, kb, cfg, "gdrive") - if err != nil { - return nil, err - } - if tok == "" { - return nil, fmt.Errorf("gdrive not configured (set GDRIVE_TOKEN or `kb oauth set --source gdrive`)") - } - return gdrive.New(tok), nil - case "gmail": - tok, err := oauthToken(ctx, kb, cfg, "gmail") - if err != nil { - return nil, err - } - if tok == "" { - return nil, fmt.Errorf("gmail not configured (set GMAIL_TOKEN or `kb oauth set --source gmail`)") - } - q := "" - if sc := cfg.Source("gmail"); sc != nil { - q = sc.Query - } - return gmail.New(tok, gmail.WithQuery(q)), nil - case "linear": - sc := cfg.Source("linear") - if sc == nil || sc.Token == "" { - return nil, fmt.Errorf("linear source not configured (set an API key via LINEAR_TOKEN)") - } - return linear.New(sc.Token), nil - case "gitlab": - sc := cfg.Source("gitlab") - if sc == nil || sc.Token == "" { - return nil, fmt.Errorf("gitlab source not configured (set an access token via GITLAB_TOKEN)") - } - projects := sc.Repos - if path != "" { // --path holds a group/project for a targeted import - projects = []string{path} - } - if len(projects) == 0 { - return nil, fmt.Errorf("gitlab needs a project: --path group/project, or sources[].repos / GITLAB_PROJECTS") - } - return gitlab.New(sc.Token, sc.BaseURL, projects), nil - case "jira": - sc := cfg.Source("jira") - if sc == nil || sc.BaseURL == "" || sc.Email == "" || sc.Token == "" { - return nil, fmt.Errorf("jira source not configured (set JIRA_BASE_URL, JIRA_EMAIL, JIRA_TOKEN)") - } - return jira.New(sc.BaseURL, sc.Email, sc.Token), nil - case "confluence": - sc := cfg.Source("confluence") - if sc == nil || sc.BaseURL == "" || sc.Email == "" || sc.Token == "" { - return nil, fmt.Errorf("confluence source not configured (set CONFLUENCE_BASE_URL, CONFLUENCE_EMAIL, CONFLUENCE_TOKEN)") - } - return confluence.New(sc.BaseURL, sc.Email, sc.Token), nil - case "markdown": - dir := path - if dir == "" { - if sc := cfg.Source("markdown"); sc != nil { - dir = sc.Path - } - } - if dir == "" { - return nil, fmt.Errorf("markdown source needs a directory (--path or sources[].path)") - } - return markdown.New(dir, markdown.WithOCR(ocrFunc(cfg))), nil - default: - return nil, fmt.Errorf("unknown source %q", source) - } + return connectors.Build(ctx, kb, cfg, source, path, ocrFunc(cfg)) } func consolidateCmd() *cobra.Command { diff --git a/cmd/http/main.go b/cmd/http/main.go index 2420974..152f45b 100644 --- a/cmd/http/main.go +++ b/cmd/http/main.go @@ -23,7 +23,9 @@ import ( "time" "github.com/programmism/brainiac/internal/applog" + "github.com/programmism/brainiac/internal/chunk" "github.com/programmism/brainiac/internal/config" + "github.com/programmism/brainiac/internal/connectors" "github.com/programmism/brainiac/internal/core" "github.com/programmism/brainiac/internal/logbuf" "github.com/programmism/brainiac/internal/plugins/anthropic" @@ -157,6 +159,13 @@ func run() error { go autoImport(shutdownCtx, c, cfg, d) } + // Optional background connector sync (#428): opt-in via SYNC_INTERVAL, re-ingests + // the configured remote connectors on a timer (the in-process form of `kb sync`). + // Off by default, so the minimal single-user stack is unchanged. + if d := cfg.SyncInterval(); d > 0 { + go autoSync(shutdownCtx, c, cfg, d) + } + logStartupBanner(cfg, writable, embedderCheck) if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { return err @@ -313,6 +322,63 @@ func autoImport(ctx context.Context, c *core.Core, cfg *config.Config, every tim } } +// autoSync re-ingests the configured remote connectors on a timer (#428) — the +// in-process equivalent of running `kb sync` periodically. markdown is skipped +// (autoImport already sweeps ./data/docs and markdown sources); each connector is +// built via the shared builder and ingested incrementally so repeated runs are +// cheap. A per-source failure is logged and the sweep continues to the next. +func autoSync(ctx context.Context, c *core.Core, cfg *config.Config, every time.Duration) { + sources := make([]string, 0, len(cfg.ConnectorSources())) + for _, s := range cfg.ConnectorSources() { + if s == "markdown" { + continue // covered by autoImport + } + sources = append(sources, s) + } + if len(sources) == 0 { + log.Printf("connector-sync enabled every %s, but no remote connectors are configured — nothing to sync", every) + return + } + run := func() { + for _, src := range sources { + if ctx.Err() != nil { + return + } + conn, err := connectors.Build(ctx, c, cfg, src, "", nil) + if err != nil { + log.Printf("connector-sync: %s: %v", src, err) + continue + } + opts := core.IngestOptions{ + Incremental: true, + PruneMissing: cfg.Ingest.PruneDeleted, + Trust: cfg.SourceTrust(src), + ChunkParams: chunk.Preset(cfg.SourceChunkPreset(src)), + } + stats, err := c.Ingest(ctx, conn, opts) + if err != nil { + log.Printf("connector-sync: %s: %v", src, err) + continue + } + if stats.Kept+stats.Queued+stats.Deleted > 0 { + log.Printf("connector-sync: %s: +%d kept, %d deleted", src, stats.Kept+stats.Queued, stats.Deleted) + } + } + } + log.Printf("connector-sync enabled every %s (%s)", every, strings.Join(sources, ", ")) + run() + t := time.NewTicker(every) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + run() + } + } +} + // ollamaChecker returns a readiness probe for the embedder backend. It hits the // Ollama tags endpoint; a failure is reported but never fatal (§11). // extractorOptions builds the optional local-LLM extractor wiring from config. diff --git a/cmd/mcp/main.go b/cmd/mcp/main.go index ab5723d..2e42e39 100644 --- a/cmd/mcp/main.go +++ b/cmd/mcp/main.go @@ -16,22 +16,13 @@ import ( "github.com/programmism/brainiac/internal/applog" "github.com/programmism/brainiac/internal/chunk" "github.com/programmism/brainiac/internal/config" + "github.com/programmism/brainiac/internal/connectors" "github.com/programmism/brainiac/internal/core" "github.com/programmism/brainiac/internal/doctext" "github.com/programmism/brainiac/internal/mcpserver" "github.com/programmism/brainiac/internal/plugins/anthropic" - "github.com/programmism/brainiac/internal/plugins/confluence" "github.com/programmism/brainiac/internal/plugins/density" - "github.com/programmism/brainiac/internal/plugins/gdrive" - "github.com/programmism/brainiac/internal/plugins/github" - "github.com/programmism/brainiac/internal/plugins/gitlab" - "github.com/programmism/brainiac/internal/plugins/gmail" - "github.com/programmism/brainiac/internal/plugins/jira" - "github.com/programmism/brainiac/internal/plugins/linear" - "github.com/programmism/brainiac/internal/plugins/markdown" - "github.com/programmism/brainiac/internal/plugins/notion" "github.com/programmism/brainiac/internal/plugins/ollama" - "github.com/programmism/brainiac/internal/plugins/slack" "github.com/programmism/brainiac/internal/store" ) @@ -173,124 +164,21 @@ func retrievalOption(cfg *config.Config) core.Option { }) } -// oauthToken resolves a connector's access token (#246): a stored, auto-refreshed -// OAuth credential if present, else the _TOKEN env value. Errors when neither -// is set. -func oauthToken(ctx context.Context, c *core.Core, source string, cfg *config.Config) (string, error) { - env := "" - if sc := cfg.Source(source); sc != nil { - env = sc.Token - } - tok, err := c.ResolveSourceToken(ctx, source, env) - if err != nil { - return "", err - } - if tok == "" { - return "", fmt.Errorf("%s is not configured (set %s_TOKEN or `kb oauth set --source %s`)", source, strings.ToUpper(source), source) - } - return tok, nil -} - -func gmailQuery(cfg *config.Config) string { - if sc := cfg.Source("gmail"); sc != nil { - return sc.Query - } - return "" -} - -// importFunc dispatches an MCP ingest request to the right connector, keeping -// the mcp/core layers plugin-agnostic. +// importFunc dispatches an MCP ingest request to the right connector via the +// shared internal/connectors builder (#428), keeping the mcp/core layers +// plugin-agnostic. An empty markdown target falls back to the container's +// /data/docs mount. func importFunc(c *core.Core, cfg *config.Config) mcpserver.ImportFunc { return func(ctx context.Context, source, target, project string) (core.IngestStats, error) { opts := core.IngestOptions{Project: project, Trust: cfg.SourceTrust(source), ChunkParams: chunk.Preset(cfg.SourceChunkPreset(source))} - switch source { - case "markdown": - dir := target - if dir == "" { - dir = "/data/docs" - } - return c.Ingest(ctx, markdown.New(dir, markdown.WithOCR(ocrFunc(cfg))), opts) - case "notion": - sc := cfg.Source("notion") - if sc == nil || sc.Token == "" { - return core.IngestStats{}, fmt.Errorf("notion is not configured (set NOTION_TOKEN)") - } - if target == "" { - return c.Ingest(ctx, notion.New(sc.Token), opts) - } - return c.Ingest(ctx, notion.NewForPages(sc.Token, []string{target}), opts) - case "slack": - sc := cfg.Source("slack") - if sc == nil || sc.Token == "" { - return core.IngestStats{}, fmt.Errorf("slack is not configured (set SLACK_TOKEN)") - } - if target == "" { - return c.Ingest(ctx, slack.New(sc.Token), opts) - } - return c.Ingest(ctx, slack.NewForChannels(sc.Token, []string{target}), opts) - case "github": - sc := cfg.Source("github") - if sc == nil || sc.Token == "" { - return core.IngestStats{}, fmt.Errorf("github is not configured (set GITHUB_TOKEN)") - } - repos := sc.Repos - if target != "" { - repos = []string{target} - } - if len(repos) == 0 { - return core.IngestStats{}, fmt.Errorf("github needs a repo: pass owner/repo as the target, or set sources[].repos / GITHUB_REPOS") - } - ghOpts := []github.Option{github.WithFiles(sc.Files)} - if sc.Discussions { - ghOpts = append(ghOpts, github.WithDiscussions()) - } - return c.Ingest(ctx, github.New(sc.Token, repos, ghOpts...), opts) - case "gdrive": - tok, err := oauthToken(ctx, c, "gdrive", cfg) - if err != nil { - return core.IngestStats{}, err - } - return c.Ingest(ctx, gdrive.New(tok), opts) - case "gmail": - tok, err := oauthToken(ctx, c, "gmail", cfg) - if err != nil { - return core.IngestStats{}, err - } - return c.Ingest(ctx, gmail.New(tok, gmail.WithQuery(gmailQuery(cfg))), opts) - case "linear": - sc := cfg.Source("linear") - if sc == nil || sc.Token == "" { - return core.IngestStats{}, fmt.Errorf("linear is not configured (set LINEAR_TOKEN)") - } - return c.Ingest(ctx, linear.New(sc.Token), opts) - case "gitlab": - sc := cfg.Source("gitlab") - if sc == nil || sc.Token == "" { - return core.IngestStats{}, fmt.Errorf("gitlab is not configured (set GITLAB_TOKEN)") - } - projects := sc.Repos - if target != "" { - projects = []string{target} - } - if len(projects) == 0 { - return core.IngestStats{}, fmt.Errorf("gitlab needs a project: pass group/project as the target, or set sources[].repos / GITLAB_PROJECTS") - } - return c.Ingest(ctx, gitlab.New(sc.Token, sc.BaseURL, projects), opts) - case "jira": - sc := cfg.Source("jira") - if sc == nil || sc.BaseURL == "" || sc.Email == "" || sc.Token == "" { - return core.IngestStats{}, fmt.Errorf("jira is not configured (set JIRA_BASE_URL, JIRA_EMAIL, JIRA_TOKEN)") - } - return c.Ingest(ctx, jira.New(sc.BaseURL, sc.Email, sc.Token), opts) - case "confluence": - sc := cfg.Source("confluence") - if sc == nil || sc.BaseURL == "" || sc.Email == "" || sc.Token == "" { - return core.IngestStats{}, fmt.Errorf("confluence is not configured (set CONFLUENCE_BASE_URL, CONFLUENCE_EMAIL, CONFLUENCE_TOKEN)") - } - return c.Ingest(ctx, confluence.New(sc.BaseURL, sc.Email, sc.Token), opts) - default: - return core.IngestStats{}, fmt.Errorf("unknown source %q (use notion, slack, github, gdrive, linear, gitlab, jira, confluence, or markdown)", source) + if source == "markdown" && target == "" { + target = "/data/docs" + } + conn, err := connectors.Build(ctx, c, cfg, source, target, ocrFunc(cfg)) + if err != nil { + return core.IngestStats{}, err } + return c.Ingest(ctx, conn, opts) } } diff --git a/internal/config/config.go b/internal/config/config.go index 7fd6a3f..a8c46e7 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -210,6 +210,12 @@ func (p PrincipalConfig) ReadNamespaces() []string { type IngestConfig struct { // Interval as a Go duration string (e.g. "60s"); empty disables auto-import. Interval string `yaml:"interval"` + // SyncInterval, a Go duration string (e.g. "15m"), enables the opt-in + // in-process background sync of the configured remote connectors (#428) — + // notion/slack/github/… are re-ingested on this timer, the same work `kb sync` + // does by hand. Empty (the default) disables it, so the minimal single-user + // stack is unchanged; markdown auto-import stays governed by Interval. + SyncInterval string `yaml:"sync_interval,omitempty"` // PruneDeleted propagates source-side deletions on auto-import (#247/#323): a // local doc removed from ./data/docs is removed from memory on the next sweep. // Off by default — the retention default (#107) keeps deleted content — so it's @@ -226,6 +232,16 @@ func (c *Config) AutoImportInterval() time.Duration { return d } +// SyncInterval returns the parsed background connector-sync interval (#428), or 0 +// if unset/invalid — 0 keeps the in-process scheduler off (opt-in). +func (c *Config) SyncInterval() time.Duration { + d, err := time.ParseDuration(c.Ingest.SyncInterval) + if err != nil { + return 0 + } + return d +} + // OCRConfig enables the opt-in OCR fallback for scanned/image-only PDFs (#356). // Off by default (pluggability): only when Enabled and an external Command is set // does an image-only PDF get OCR'd. Command is invoked as ` @@ -628,6 +644,9 @@ func (c *Config) applyEnvOverrides() { if v := os.Getenv("INGEST_INTERVAL"); v != "" { c.Ingest.Interval = v } + if v := os.Getenv("SYNC_INTERVAL"); v != "" { + c.Ingest.SyncInterval = v + } if v := os.Getenv("INGEST_PRUNE_DELETED"); v != "" { c.Ingest.PruneDeleted = v == "true" || v == "1" } diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 4694816..b58b794 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -823,6 +823,33 @@ func TestAutoImportInterval(t *testing.T) { } } +func TestSyncInterval(t *testing.T) { + c := Default() + if c.SyncInterval() != 0 { + t.Error("empty sync interval should be 0 (disabled by default)") + } + c.Ingest.SyncInterval = "15m" + if c.SyncInterval() != 15*time.Minute { + t.Errorf("got %v, want 15m", c.SyncInterval()) + } + c.Ingest.SyncInterval = "garbage" + if c.SyncInterval() != 0 { + t.Error("invalid sync interval should be 0") + } +} + +func TestSyncIntervalFromEnv(t *testing.T) { + t.Setenv("DATABASE_URL", "postgres://env-dsn") + t.Setenv("SYNC_INTERVAL", "30m") + c, err := Load(filepath.Join(t.TempDir(), "none.yaml")) + if err != nil { + t.Fatal(err) + } + if c.SyncInterval() != 30*time.Minute { + t.Errorf("SYNC_INTERVAL env: got %v, want 30m", c.SyncInterval()) + } +} + func TestGitHubTokenAndReposAutoCreateSource(t *testing.T) { t.Setenv("NOTION_TOKEN", "") t.Setenv("SLACK_TOKEN", "") diff --git a/internal/connectors/connectors.go b/internal/connectors/connectors.go new file mode 100644 index 0000000..982725f --- /dev/null +++ b/internal/connectors/connectors.go @@ -0,0 +1,146 @@ +// Package connectors builds a plugins.SourceConnector for a configured source +// type — the one place the cmd adapters (CLI import, MCP ingest, the HTTP +// background sync) share so connector construction isn't duplicated (#428). OAuth +// access tokens are resolved (and auto-refreshed) via the core (#246); the OCR +// fallback is passed in by the caller so the shell-out stays in the cmd layer. +package connectors + +import ( + "context" + "fmt" + + "github.com/programmism/brainiac/internal/config" + "github.com/programmism/brainiac/internal/core" + "github.com/programmism/brainiac/internal/doctext" + "github.com/programmism/brainiac/internal/plugins" + "github.com/programmism/brainiac/internal/plugins/confluence" + "github.com/programmism/brainiac/internal/plugins/gdrive" + "github.com/programmism/brainiac/internal/plugins/github" + "github.com/programmism/brainiac/internal/plugins/gitlab" + "github.com/programmism/brainiac/internal/plugins/gmail" + "github.com/programmism/brainiac/internal/plugins/jira" + "github.com/programmism/brainiac/internal/plugins/linear" + "github.com/programmism/brainiac/internal/plugins/markdown" + "github.com/programmism/brainiac/internal/plugins/notion" + "github.com/programmism/brainiac/internal/plugins/slack" +) + +// Build constructs the connector for a source type. path is an optional targeted +// import (a page/channel id, an owner/repo, or a markdown dir); ocr is the optional +// OCR fallback for the markdown connector (nil = off). Returns an error when the +// source isn't configured. +func Build(ctx context.Context, kb *core.Core, cfg *config.Config, source, path string, ocr doctext.OCRFunc) (plugins.SourceConnector, error) { + switch source { + case "notion": + sc := cfg.Source("notion") + if sc == nil || sc.Token == "" { + return nil, fmt.Errorf("notion source not configured (set NOTION_TOKEN or config.yaml)") + } + if path != "" { + return notion.NewForPages(sc.Token, []string{path}), nil + } + return notion.New(sc.Token), nil + case "slack": + sc := cfg.Source("slack") + if sc == nil || sc.Token == "" { + return nil, fmt.Errorf("slack source not configured (set SLACK_TOKEN or config.yaml)") + } + if path != "" { + return slack.NewForChannels(sc.Token, []string{path}), nil + } + return slack.New(sc.Token), nil + case "github": + sc := cfg.Source("github") + if sc == nil || sc.Token == "" { + return nil, fmt.Errorf("github source not configured (set GITHUB_TOKEN or config.yaml)") + } + repos := sc.Repos + if path != "" { + repos = []string{path} + } + if len(repos) == 0 { + return nil, fmt.Errorf("github needs a repo: a target owner/repo, or sources[].repos / GITHUB_REPOS") + } + ghOpts := []github.Option{github.WithFiles(sc.Files)} + if sc.Discussions { + ghOpts = append(ghOpts, github.WithDiscussions()) + } + return github.New(sc.Token, repos, ghOpts...), nil + case "linear": + sc := cfg.Source("linear") + if sc == nil || sc.Token == "" { + return nil, fmt.Errorf("linear source not configured (set an API key via LINEAR_TOKEN)") + } + return linear.New(sc.Token), nil + case "gitlab": + sc := cfg.Source("gitlab") + if sc == nil || sc.Token == "" { + return nil, fmt.Errorf("gitlab source not configured (set GITLAB_TOKEN)") + } + projects := sc.Repos + if path != "" { + projects = []string{path} + } + if len(projects) == 0 { + return nil, fmt.Errorf("gitlab needs a project: a target group/project, or sources[].repos / GITLAB_PROJECTS") + } + return gitlab.New(sc.Token, sc.BaseURL, projects), nil + case "jira": + sc := cfg.Source("jira") + if sc == nil || sc.Token == "" || sc.BaseURL == "" || sc.Email == "" { + return nil, fmt.Errorf("jira source not configured (set JIRA_BASE_URL, JIRA_EMAIL, JIRA_TOKEN)") + } + return jira.New(sc.BaseURL, sc.Email, sc.Token), nil + case "confluence": + sc := cfg.Source("confluence") + if sc == nil || sc.Token == "" || sc.BaseURL == "" || sc.Email == "" { + return nil, fmt.Errorf("confluence source not configured (set CONFLUENCE_BASE_URL, CONFLUENCE_EMAIL, CONFLUENCE_TOKEN)") + } + return confluence.New(sc.BaseURL, sc.Email, sc.Token), nil + case "gdrive": + tok, err := oauthToken(ctx, kb, cfg, "gdrive") + if err != nil { + return nil, err + } + if tok == "" { + return nil, fmt.Errorf("gdrive not configured (set GDRIVE_TOKEN or `kb oauth set --source gdrive`)") + } + return gdrive.New(tok), nil + case "gmail": + tok, err := oauthToken(ctx, kb, cfg, "gmail") + if err != nil { + return nil, err + } + if tok == "" { + return nil, fmt.Errorf("gmail not configured (set GMAIL_TOKEN or `kb oauth set --source gmail`)") + } + q := "" + if sc := cfg.Source("gmail"); sc != nil { + q = sc.Query + } + return gmail.New(tok, gmail.WithQuery(q)), nil + case "markdown": + dir := path + if dir == "" { + if sc := cfg.Source("markdown"); sc != nil { + dir = sc.Path + } + } + if dir == "" { + return nil, fmt.Errorf("markdown source needs a directory (a target path or sources[].path)") + } + return markdown.New(dir, markdown.WithOCR(ocr)), nil + default: + return nil, fmt.Errorf("unknown source %q", source) + } +} + +// oauthToken resolves a connector's access token (#246): a stored, auto-refreshed +// OAuth credential if present, else the source's _TOKEN env value. +func oauthToken(ctx context.Context, kb *core.Core, cfg *config.Config, source string) (string, error) { + env := "" + if sc := cfg.Source(source); sc != nil { + env = sc.Token + } + return kb.ResolveSourceToken(ctx, source, env) +} diff --git a/internal/connectors/connectors_test.go b/internal/connectors/connectors_test.go new file mode 100644 index 0000000..051ab21 --- /dev/null +++ b/internal/connectors/connectors_test.go @@ -0,0 +1,64 @@ +package connectors + +import ( + "context" + "testing" + + "github.com/programmism/brainiac/internal/config" +) + +// TestBuildConstructsConfiguredSources checks the shared builder (#428) returns a +// connector for configured sources and a clear error otherwise. Non-OAuth sources +// don't touch the core, so a nil *core.Core is fine here (OAuth resolution for +// gdrive/gmail is exercised by the core's own tests). +func TestBuildConstructsConfiguredSources(t *testing.T) { + ctx := context.Background() + dir := t.TempDir() + + cfg := &config.Config{Sources: []config.SourceConfig{ + {Type: "notion", Token: "ntn_x"}, + {Type: "github", Token: "ghp_x", Repos: []string{"o/r"}}, + {Type: "markdown", Path: dir}, + }} + + for _, tc := range []struct { + name string + source string + path string + wantErr bool + }{ + {"notion configured", "notion", "", false}, + {"github configured", "github", "", false}, + {"github via path", "github", "owner/repo", false}, + {"markdown configured", "markdown", "", false}, + {"markdown via path", "markdown", dir, false}, + {"slack missing token", "slack", "", true}, + {"markdown no dir", "markdown", "", false}, // has sources[].path + {"unknown source", "banana", "", true}, + } { + t.Run(tc.name, func(t *testing.T) { + conn, err := Build(ctx, nil, cfg, tc.source, tc.path, nil) + if tc.wantErr { + if err == nil { + t.Fatalf("Build(%q) = nil error, want error", tc.source) + } + return + } + if err != nil { + t.Fatalf("Build(%q) = %v, want nil", tc.source, err) + } + if conn == nil { + t.Fatalf("Build(%q) = nil connector", tc.source) + } + }) + } +} + +// TestBuildMarkdownNeedsDir verifies markdown with neither a path nor a configured +// directory is a configuration error (not a silent no-op). +func TestBuildMarkdownNeedsDir(t *testing.T) { + cfg := &config.Config{} + if _, err := Build(context.Background(), nil, cfg, "markdown", "", nil); err == nil { + t.Fatal("markdown with no dir: want error, got nil") + } +}