diff --git a/README.md b/README.md index aa429852..19ebfc13 100644 --- a/README.md +++ b/README.md @@ -49,13 +49,19 @@ background stack. All configuration lives in `docker/env/*.local.env` (created by `make setup`, gitignored). The defaults work out of the box; integrations (GitHub app, -Slack app, AI assistant, email) are optional and stay disabled until their -variables are set. +Slack app, AI assistant, email, changedetection.io, Meilisearch) are +optional and stay disabled until their variables are set. The frontend +learns what a deployment offers from `GET /core/api/capabilities`, which +reports a boolean per service (`github`, `slack`, `assistant`, +`changedetection`, `search`) so unavailable features are not rendered at +all. ## AI assistant -The assistant is not tied to any one vendor. Pick a provider and a model -in `docker/env/core.local.env`: +The assistant is disabled by default: with `ASSISTANT_PROVIDER` empty the +server boots without a model and the in-app chat is unavailable (the MCP +server below keeps working). It is not tied to any one vendor — enable it +by picking a provider and a model in `docker/env/core.local.env`: ```sh OXYNOTE_CORE_ASSISTANT_PROVIDER=claude diff --git a/docker/env/core.example.env b/docker/env/core.example.env index b26f6f28..cb984b2b 100644 --- a/docker/env/core.example.env +++ b/docker/env/core.example.env @@ -5,6 +5,8 @@ # infrastructure wiring (compose service hostnames) OXYNOTE_CORE_DB_DSN=postgresql://devuser:devpass@postgres/devdb?sslmode=disable +# an empty DSN disables search: documents are not indexed, the search +# endpoint refuses, and the assistant loses its search tool. OXYNOTE_CORE_MEILISEARCH_DSN=http://meilisearch:7700 OXYNOTE_CORE_MEILISEARCH_MASTER_KEY=devmasterkey OXYNOTE_CORE_VALKEY_NETWORK=tcp @@ -17,6 +19,8 @@ OXYNOTE_CORE_SERVER_AUTH_BETTER_AUTH_URL=http://auth-realtime:8081/api/auth/get- OXYNOTE_CORE_SERVER_MCP_SESSION_URL=http://auth-realtime:8081/api/internal/mcp/session OXYNOTE_CORE_AUTH_REALTIME_URL=http://auth-realtime:8081 OXYNOTE_CORE_SERVER_DEMO_PROMETHEUS_URL=http://prometheus:9090 +# an empty URL disables changedetection.io: url-watcher hooks cannot be +# created and existing ones are skipped. OXYNOTE_CORE_CHANGEDETECTION_API_URL=http://changedetection:5000 # auto-populated in core.local.env by make run/start/dev (see # docker/sync-changedetection-key.sh) — no need to fill in manually. @@ -54,16 +58,18 @@ OXYNOTE_CORE_SLACK_CLIENT_SECRET= OXYNOTE_CORE_SLACK_SIGNATURE_SECRET= OXYNOTE_CORE_SLACK_REDIRECT_URL=http://localhost:8080/slack OXYNOTE_CORE_SLACK_INSTALLATION_SIGNING_SECRET= -# AI assistant. PROVIDER selects the vendor: claude, openai, gemini, -# ollama, or openrouter. MODEL is passed to the provider verbatim, so use -# that vendor's own identifier. Both are required; core refuses to start -# without them. +# AI assistant. An empty PROVIDER disables the assistant: core boots, +# the chat endpoint refuses, and the MCP surface keeps working. To +# enable it, set PROVIDER to the vendor (claude, openai, gemini, ollama, +# or openrouter), MODEL to that vendor's own model identifier, and the +# credentials the vendor needs (an API key for every provider except +# ollama); a set provider with missing pieces is a boot error. # # The assistant is entirely tool-driven, so the model must be reliable at # calling tools against a large schema. Known-good choices per provider # are listed in the README; small local models generally are not. -OXYNOTE_CORE_ASSISTANT_PROVIDER=claude -OXYNOTE_CORE_ASSISTANT_MODEL=claude-opus-4-6 +OXYNOTE_CORE_ASSISTANT_PROVIDER= +OXYNOTE_CORE_ASSISTANT_MODEL= OXYNOTE_CORE_ASSISTANT_API_KEY= # optional. Overrides the provider endpoint. Required for ollama, and the # way to reach any OpenAI-compatible server with PROVIDER=openai. diff --git a/server/CLAUDE.md b/server/CLAUDE.md index 0f1a8943..2abd34d4 100644 --- a/server/CLAUDE.md +++ b/server/CLAUDE.md @@ -61,7 +61,7 @@ So from a frontend's point of view: auth + realtime is `:8080/auth-realtime/...` `server/core/internal/server/router.go` is the canonical map. Three layers: -- `/api/...` (public): auth middleware (`internal/server/internal/auth`) validates sessions by calling auth-realtime's `/api/auth/get-session` (configured via `SERVER_AUTH_BETTER_AUTH_URL`). Routes for documents, branches, comments, reviewers, hooks, files, GitHub, Slack, notifications, data-sources, AI chat. +- `/api/...` (public): auth middleware (`internal/server/internal/auth`) validates sessions by calling auth-realtime's `/api/auth/get-session` (configured via `SERVER_AUTH_BETTER_AUTH_URL`). Routes for documents, branches, comments, reviewers, hooks, files, GitHub, Slack, notifications, data-sources, AI chat. `GET /api/capabilities` is the frontend's capability signal: one boolean per optional service (`github`, `slack`, `assistant`, `changedetection`, `search`), snapshotted at boot from each client's/manager's `Configured()`. - `/api/x/...` (internal): no auth at all — reverse proxy must firewall. Used by auth-realtime to fetch/store branch content (`/x/documents/{id}/branches`, `/x/documents/{id}/branch/{branchId}`), trigger emails, and initialize or tear down orgs. - `/api/apps/...` (public, sessionless): where GitHub and Slack deliver. `POST /apps/github/events` and `POST /apps/slack/{events,commands,slash}` are gated by the provider's request signature; `GET /apps/slack/install` completes the direct-install OAuth exchange. These must stay outside `/api/x` — third parties reach core through the same front door as browsers, and the proxy 403s the internal subtree. - `/api/mcp` (public, bearer-authed): the MCP surface (`internal/server/internal/mcp`), a streamable HTTP MCP server bridging the assistant's ungated tool registry (`tools.Set.Entries`) plus documents-as-resources. Bearer tokens are JWTs issued by auth-realtime's `@better-auth/mcp` OAuth provider; core validates each request against auth-realtime's internal `GET /api/internal/mcp/session` (JWKS verify + consent-row check, so revoking a client 401s immediately) and scopes the tool list by the token's `documents:read` / `documents:write` / `data-sources:read` scopes. Tokens are org-bound via an `org_id` claim minted at issuance. `SERVER_MCP_SESSION_URL` and `SERVER_MCP_RESOURCE_URL` are required env; the Caddyfile routes `/.well-known/oauth-*` and `/api/auth/*` at the front door to auth-realtime for OAuth discovery. @@ -92,7 +92,7 @@ Uploaded images live in MinIO under `organizations/{org}/documents/{doc}/files/{ - **Ordering is inverted between create and delete.** Creating writes the row before the object (upload, duplication); deleting removes the object before the row. Hooks follow the same rule from outside the transaction: `copyHooksToBranch` runs after the fork or merge commits, because `hook.NewHook` creates the watcher as a side effect and a rollback cannot take that back; a failed insert tears the just-created watcher down again. Both leave at most a row without an object, which the sweep resolves — an object without a row is invisible to every cleanup path. - **Duplication copies, never shares.** `Document.Duplicate` regenerates every block uid, rewrites image `src` paths to the new document (swapping only the path, never the host), and returns the old→new id map the handler uses to copy each object server-side into the duplicate's own folder. Files are therefore owned by exactly one document. - **Search jobs apply in order, per document.** A diff is a delta against the state its predecessors left, so the manager holds back every later diff of a document whose earlier one failed, rather than replaying it on top of newer work; an organization removal is global and holds everything after it. Without this a failed add replayed after a delete re-inserts the blocks of a document that no longer exists, and no later job would ever remove them. -- **Search index entries are removed by the delete itself.** Only default-branch blocks are indexed, keyed by block uid with a `documentId` field. `agent.DeleteDocument` collects the subtree ids before the row goes away and queues a search job carrying them, so the manager clears the index by `documentId` filter — the descendants the cascade destroys have no other trace to clean up by. Deletion paths therefore never queue the removal themselves. +- **Search index entries are removed from the ids the delete reports.** Only default-branch blocks are indexed, keyed by block uid with a `documentId` field. `agent.DeleteDocument` collects the subtree ids before the row goes away and returns them, and the deletion paths queue a search job carrying them through `search.Jobs` in the same transaction, so the manager clears the index by `documentId` filter — the descendants the cascade destroys have no other trace to clean up by. - Snapshots pin the files they reference, so `DB_MAX_DOCUMENT_CHANGELOGS` (count, trimmed on insert) and `DB_DOCUMENT_CHANGELOG_RETENTION` (age, trimmed in the sweep) also decide how long a removed image survives. The ProseMirror schema is defined in `server/auth-realtime/src/schema/` (one file per block kind, `index.ts` aggregates). The Go-side mirror is `server/core/internal/document/node.go` (`RootBlock`, `Block`, `Mark`). @@ -116,7 +116,7 @@ Env lives in `docker/env/`: committed `*.example.env` templates list **every** v - core vars are prefixed `OXYNOTE_CORE_`; `buildinfo.Getenv("FOO")` reads `OXYNOTE_CORE_FOO`. - auth-realtime vars are prefixed `OXYNOTE_AUTH_REALTIME_`. -- All integrations are optional. The GitHub App is enabled by setting `OXYNOTE_CORE_GITHUB_APP_ID` and dropping the app's private key into `docker/github/` (mounted into the core container); an empty `OXYNOTE_CORE_GITHUB_APP_ID` disables it — core boots, GitHub routes respond `github.not_configured` (the always-200 `GET /api/github` status endpoint reports `configured: false` as the frontend's capability signal), and github-tracking hooks are skipped. A set app ID with a missing/unreadable key is a boot error. Slack works the same way keyed on `OXYNOTE_CORE_SLACK_CLIENT_ID` (`slack.not_configured`, `GET /api/slack` reports `configured`); a set client ID with other `SLACK_*` values missing is a boot error. The assistant needs `ASSISTANT_PROVIDER` and `ASSISTANT_MODEL`, and an `ASSISTANT_API_KEY` for every provider except ollama; missing or unsupported values are a boot error, since a chat that cannot reach a model is worse than a refusal to start. Provider-specific credentials (`ASSISTANT_BEDROCK_*`, `ASSISTANT_VERTEX_*`, `ASSISTANT_AZURE_API_VERSION`) are only read when the selected provider uses them. Email is keyed on `OXYNOTE_CORE_EMAIL_SMTP_HOST`: when empty, core boots and each would-be email is logged instead of sent; with a set host, an invalid `EMAIL_SMTP_PORT` or `EMAIL_SMTP_TLS` (`none` | `starttls` | `tls`) is a boot error. The dev stack presets the mailpit container as the SMTP target (`mailpit:1025`, plaintext, no auth; web UI on host `:8025`) — the four HTML templates are embedded in the core binary from `server/core/internal/email/templates/`. +- All integrations are optional, each keyed on one variable; a set key with the rest of its group missing or invalid is a boot error, and `GET /api/capabilities` reports what ended up enabled. The GitHub App is enabled by setting `OXYNOTE_CORE_GITHUB_APP_ID` and dropping the app's private key into `docker/github/` (mounted into the core container); an empty `OXYNOTE_CORE_GITHUB_APP_ID` disables it — core boots, GitHub routes respond `github.not_configured` (the always-200 `GET /api/github` status endpoint also reports `configured: false`), and github-tracking hooks are skipped. A set app ID with a missing/unreadable key is a boot error. Slack works the same way keyed on `OXYNOTE_CORE_SLACK_CLIENT_ID` (`slack.not_configured`, `GET /api/slack` reports `configured`); a set client ID with other `SLACK_*` values missing is a boot error. The assistant is keyed on `ASSISTANT_PROVIDER`: empty disables it (`GET /api/ai/chat` responds `assistant.not_configured` before any websocket upgrade, no model objects are built, the MCP surface keeps working, and the other `ASSISTANT_*` values are ignored); a set provider needs `ASSISTANT_MODEL` and an `ASSISTANT_API_KEY` for every provider except ollama, and missing or unsupported values are a boot error. Provider-specific credentials (`ASSISTANT_BEDROCK_*`, `ASSISTANT_VERTEX_*`, `ASSISTANT_AZURE_API_VERSION`) are only read when the selected provider uses them. Search is keyed on `MEILISEARCH_DSN`: empty disables it — `GET /api/documents/search` responds `search.not_configured`, every enqueue goes through `search.Jobs` (`internal/search/jobs.go`) which drops jobs on a disabled deployment, the search-job manager is not started, and the `search_documents` tool is left out of the assistant/MCP registry; a set DSN that cannot be reached is a boot error. Changedetection.io is keyed on `CHANGEDETECTION_API_URL`: empty disables it — creating, updating or resetting a url-watcher hook responds `changedetection.not_configured`, existing url-watcher hooks are skipped by the processor and by fork/merge hook copies, and deleting one still removes the row (`webchange.Client.DeleteWatcher` is a no-op when unconfigured); `CHANGEDETECTION_API_KEY` may stay empty either way. Email is keyed on `OXYNOTE_CORE_EMAIL_SMTP_HOST`: when empty, core boots and each would-be email is logged instead of sent; with a set host, an invalid `EMAIL_SMTP_PORT` or `EMAIL_SMTP_TLS` (`none` | `starttls` | `tls`) is a boot error. The dev stack presets the mailpit container as the SMTP target (`mailpit:1025`, plaintext, no auth; web UI on host `:8025`) — the four HTML templates are embedded in the core binary from `server/core/internal/email/templates/`. ## Assistant prompt diff --git a/server/core/cmd/core/main.go b/server/core/cmd/core/main.go index d53e3a0e..37f277f2 100644 --- a/server/core/cmd/core/main.go +++ b/server/core/cmd/core/main.go @@ -194,18 +194,26 @@ func main() { //nolint:maintidx // main performs linear wiring of all components termCtx, termCancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer termCancel() - searchClient, err := search.NewClient( - termCtx, - meilisearch.New( - buildinfo.Getenv("MEILISEARCH_DSN"), + // an empty MEILISEARCH_DSN means search is disabled: nothing is + // indexed and the search surfaces refuse. A set DSN that cannot be + // reached stays a boot error inside NewClient. + var meiliMan meilisearch.ServiceManager + + if dsn := buildinfo.Getenv("MEILISEARCH_DSN"); dsn != "" { + meiliMan = meilisearch.New( + dsn, meilisearch.WithAPIKey(buildinfo.Getenv("MEILISEARCH_MASTER_KEY")), - ), - ) + ) + } + + searchClient, err := search.NewClient(termCtx, meiliMan) if err != nil { fail(log, closers, "cannot create search gateway client", err) return } + searchJobs := search.NewJobs(searchClient.Configured()) + storageClient, err := storage.NewClient( termCtx, storage.Options{ @@ -246,22 +254,34 @@ func main() { //nolint:maintidx // main performs linear wiring of all components buildinfo.Getenv("AUTH_REALTIME_URL"), ) - assistantOpts, err := assistantProviderOptions() - if err != nil { - fail(log, closers, "cannot read the assistant configuration", err) - return - } + // an empty ASSISTANT_PROVIDER means the assistant is disabled: no + // model is built and the chat surface refuses. The remaining + // ASSISTANT_* values are ignored, matching the other integrations. A + // set provider still validates everything it needs and fails boot. + var ( + chatModel model.ToolCallingChatModel + summaryModel model.ToolCallingChatModel + assistantOpts provider.Options + ) - chatModel, err := provider.New(termCtx, assistantOpts) - if err != nil { - fail(log, closers, "cannot create the assistant chat model", err) - return - } + if buildinfo.Getenv("ASSISTANT_PROVIDER") != "" { + assistantOpts, err = assistantProviderOptions() + if err != nil { + fail(log, closers, "cannot read the assistant configuration", err) + return + } - summaryModel, err := assistantSummaryModel(termCtx, assistantOpts, chatModel) - if err != nil { - fail(log, closers, "cannot create the assistant summarization model", err) - return + chatModel, err = provider.New(termCtx, assistantOpts) + if err != nil { + fail(log, closers, "cannot create the assistant chat model", err) + return + } + + summaryModel, err = assistantSummaryModel(termCtx, assistantOpts, chatModel) + if err != nil { + fail(log, closers, "cannot create the assistant summarization model", err) + return + } } datasourceMan := datasource.NewManager(log, dbc) @@ -275,10 +295,20 @@ func main() { //nolint:maintidx // main performs linear wiring of all components metrics, editClient, searchClient, + searchJobs, datasourceMan, string(assistantOpts.Provider), ) + // a disabled integration is a deliberate state, but it is also the + // first thing an operator looks for when a feature is missing from + // the product, so each one announces itself at boot. + warnDisabled(log, githubMan.Configured(), "github app integration is disabled") + warnDisabled(log, slackMan.Configured(), "slack app integration is disabled") + warnDisabled(log, assistantMan.Configured(), "assistant is disabled") + warnDisabled(log, searchClient.Configured(), "search is disabled") + warnDisabled(log, webchangeClient.Configured(), "changedetection integration is disabled") + srv, err := server.NewServer( log, server.Options{ @@ -303,6 +333,7 @@ func main() { //nolint:maintidx // main performs linear wiring of all components slackMan, webchangeClient, searchClient, + searchJobs, notifMan, emailSender, http.DefaultClient, @@ -313,7 +344,6 @@ func main() { //nolint:maintidx // main performs linear wiring of all components } hooksMan := hookMan.NewManager(log, dbc, githubMan, webchangeClient, notifMan) - searchManager := searchMan.NewManager(log, dbc, searchClient) filesMan := fileMan.NewManager(log, dbc, storageClient, fileMan.Options{ ChangelogRetention: documentChangelogRetention, }) @@ -343,9 +373,14 @@ func main() { //nolint:maintidx // main performs linear wiring of all components }) backgroundSupv.Go(hooksMan.Start) - backgroundSupv.Go(searchManager.Start) backgroundSupv.Go(filesMan.Start) + // without search there are no queued jobs to drain, so the manager + // is not started at all. + if searchClient.Configured() { + backgroundSupv.Go(searchMan.NewManager(log, dbc, searchClient).Start) + } + <-termCtx.Done() // the managers query the database and redis, so let them finish their @@ -471,6 +506,15 @@ func parseDurationEnv(name string, def time.Duration) (time.Duration, error) { return res, nil } +// warnDisabled announces a disabled integration at boot. +func warnDisabled(log *slog.Logger, configured bool, msg string) { + if configured { + return + } + + log.Warn(msg) +} + // parseOrigins splits the configured allowed-origin list. An empty value // yields no origins at all, which is what leaves origin checking disabled: // strings.Split would instead return a one-element slice holding "", which diff --git a/server/core/cmd/core/main_test.go b/server/core/cmd/core/main_test.go index 1c4a8757..74fb8465 100644 --- a/server/core/cmd/core/main_test.go +++ b/server/core/cmd/core/main_test.go @@ -92,3 +92,19 @@ func Test_fail(t *testing.T) { }) } } + +func Test_warnDisabled(t *testing.T) { + t.Parallel() + + // a configured integration announces nothing. + var buf bytes.Buffer + + log := slog.New(slog.NewTextHandler(&buf, nil)) + + warnDisabled(log, true, "search is disabled") + assert.Empty(t, buf.String()) + + // a disabled one warns. + warnDisabled(log, false, "search is disabled") + assert.Contains(t, buf.String(), "search is disabled") +} diff --git a/server/core/internal/apps/webchange/client.go b/server/core/internal/apps/webchange/client.go index 17ec6789..3ba3ab69 100644 --- a/server/core/internal/apps/webchange/client.go +++ b/server/core/internal/apps/webchange/client.go @@ -10,6 +10,8 @@ import ( "io" "net/http" "time" + + "github.com/oxynote/oxynote/server/core/pkg/errutil" ) // ErrWatcherNotFound is returned when the watcher no longer exists on the @@ -17,6 +19,10 @@ import ( // again rather than failing forever. var ErrWatcherNotFound = errors.New("watcher not found") +// ErrNotConfigured is returned when the changedetection.io integration is not +// configured on this deployment. +var ErrNotConfigured = errutil.New(http.StatusConflict, "changedetection.not_configured", "changedetection is not configured") + // _requestTimeout bounds a single changedetection.io request. The callers // are hook processors running under the periodic executor's long-lived // context, so without a deadline of its own a hung connection would stall @@ -39,8 +45,18 @@ func NewClient(baseURL, apiKey string) *Client { } } +// Configured reports whether the changedetection.io integration is +// configured on this deployment. An empty base URL disables it. +func (c *Client) Configured() bool { + return c.baseURL != "" +} + // FetchWatcher fetches the watch with the given UUID. func (c *Client) FetchWatcher(ctx context.Context, uuid string) (*Watch, error) { + if !c.Configured() { + return nil, ErrNotConfigured + } + url := c.baseURL + "/api/v1/watch/" + uuid req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, http.NoBody) @@ -91,6 +107,10 @@ func (c *Client) FetchWatcher(ctx context.Context, uuid string) (*Watch, error) // CreateWatcher creates a new watch and returns its uuid. func (c *Client) CreateWatcher(ctx context.Context, url string) (string, error) { + if !c.Configured() { + return "", ErrNotConfigured + } + body, err := json.Marshal(watchRequest{ URL: url, FetchBackend: _fetchBackendWebDriver, @@ -133,6 +153,10 @@ func (c *Client) CreateWatcher(ctx context.Context, url string) (string, error) // UpdateWatcher updates an existing watch. func (c *Client) UpdateWatcher(ctx context.Context, uuid, url string) error { + if !c.Configured() { + return ErrNotConfigured + } + body, err := json.Marshal(watchRequest{ URL: url, FetchBackend: _fetchBackendWebDriver, @@ -165,8 +189,14 @@ func (c *Client) UpdateWatcher(ctx context.Context, uuid, url string) error { return nil } -// DeleteWatcher deletes the watch with the given UUID. +// DeleteWatcher deletes the watch with the given UUID. On an +// unconfigured deployment there is nothing to tear down, so the delete +// succeeds as a no-op rather than blocking whatever removal asked for it. func (c *Client) DeleteWatcher(ctx context.Context, uuid string) error { + if !c.Configured() { + return nil + } + req, err := http.NewRequestWithContext(ctx, http.MethodDelete, c.baseURL+"/api/v1/watch/"+uuid, http.NoBody) if err != nil { return fmt.Errorf("failed to create request: %w", err) diff --git a/server/core/internal/apps/webchange/client_test.go b/server/core/internal/apps/webchange/client_test.go index 0ab448ed..274b816c 100644 --- a/server/core/internal/apps/webchange/client_test.go +++ b/server/core/internal/apps/webchange/client_test.go @@ -59,6 +59,36 @@ func Test_NewClient(t *testing.T) { assert.Equal(t, _requestTimeout, c.client.Timeout) } +func Test_Client_Configured(t *testing.T) { + t.Parallel() + + assert.True(t, NewClient("http://test.com", "key").Configured()) + assert.False(t, NewClient("", "key").Configured()) +} + +// Test_Client_unconfigured covers every method refusing to act without a +// base URL, so no case below has to repeat the guard. +func Test_Client_unconfigured(t *testing.T) { + t.Parallel() + + c := NewClient("", "") + ctx := context.Background() + + watch, err := c.FetchWatcher(ctx, "uuid-1") + assert.ErrorIs(t, err, ErrNotConfigured) + assert.Nil(t, watch) + + id, err := c.CreateWatcher(ctx, "https://example.com") + assert.ErrorIs(t, err, ErrNotConfigured) + assert.Empty(t, id) + + assert.ErrorIs(t, c.UpdateWatcher(ctx, "uuid-1", "https://example.com"), ErrNotConfigured) + + // deleting is the exception: there is nothing to tear down, so the + // caller's removal proceeds instead of failing. + assert.NoError(t, c.DeleteWatcher(ctx, "uuid-1")) +} + func Test_Client_FetchWatcher(t *testing.T) { t.Parallel() diff --git a/server/core/internal/assistant/manager.go b/server/core/internal/assistant/manager.go index 77ab6a53..264f9b2c 100644 --- a/server/core/internal/assistant/manager.go +++ b/server/core/internal/assistant/manager.go @@ -15,6 +15,7 @@ import ( "fmt" "io" "log/slog" + "net/http" "time" "github.com/cloudwego/eino/components/model" @@ -23,10 +24,16 @@ import ( "github.com/oxynote/oxynote/server/core/internal/assistant/persist" "github.com/oxynote/oxynote/server/core/internal/assistant/protocol" "github.com/oxynote/oxynote/server/core/internal/assistant/tools" + "github.com/oxynote/oxynote/server/core/internal/search" + "github.com/oxynote/oxynote/server/core/pkg/errutil" "github.com/oxynote/oxynote/server/core/pkg/metricutil" "github.com/oxynote/oxynote/server/core/pkg/redkit" ) +// ErrNotConfigured is returned when the assistant is not configured on this +// deployment. +var ErrNotConfigured = errutil.New(http.StatusConflict, "assistant.not_configured", "assistant is not configured") + // _sessionExpiration is how long Redis retains an idle conversation. // Conversations resume from this history, and a turn paused on a // confirmation resumes from its checkpoint, for this long. @@ -37,6 +44,7 @@ type Manager struct { log *slog.Logger db tools.DB search tools.Searcher + jobs *search.Jobs runners tools.DataSourceRunners model model.ToolCallingChatModel summary model.ToolCallingChatModel @@ -55,7 +63,8 @@ type Manager struct { // the provider package from the operator's configuration; summaryModel // backs context summarisation and may be the same model. The editClient // is the edit pipe to the Node hocuspocus service; the search client -// backs the search_documents tool; providerName labels token metrics so +// backs the search_documents tool and searchJobs is the queue document +// writes announce themselves to; providerName labels token metrics so // usage stays readable across a provider change; the tree notifier // broadcasts sidebar refresh events after document tree mutations and is // wired post-construction via SetTreeNotifier because the document @@ -68,7 +77,8 @@ func NewManager( summaryModel model.ToolCallingChatModel, fc metricutil.Factory, editClient tools.EditApplier, - search tools.Searcher, + searcher tools.Searcher, + searchJobs *search.Jobs, runners tools.DataSourceRunners, providerName string, ) *Manager { @@ -86,7 +96,8 @@ func NewManager( return &Manager{ log: log, db: db, - search: search, + search: searcher, + jobs: searchJobs, runners: runners, model: chatModel, summary: summaryModel, @@ -107,6 +118,13 @@ func NewManager( } } +// Configured reports whether the assistant is configured on this +// deployment: a manager built without a chat model has nothing to run a +// conversation on. +func (m *Manager) Configured() bool { + return m.model != nil +} + // SetTreeNotifier wires the tree-change notifier the assistant uses to // broadcast sidebar refresh events after document tree mutations. Call // this once during startup, after the document handler that satisfies @@ -124,6 +142,7 @@ func (m *Manager) ToolSet(orgID, userID string) *tools.Set { m.log, m.db, m.search, + m.jobs, m.runners, m.applier, m.tree, diff --git a/server/core/internal/assistant/manager_test.go b/server/core/internal/assistant/manager_test.go index 2dd91984..5902a7d5 100644 --- a/server/core/internal/assistant/manager_test.go +++ b/server/core/internal/assistant/manager_test.go @@ -15,6 +15,7 @@ import ( persistMock "github.com/oxynote/oxynote/server/core/internal/assistant/persist/_mock" protocolMock "github.com/oxynote/oxynote/server/core/internal/assistant/protocol/_mock" toolsMock "github.com/oxynote/oxynote/server/core/internal/assistant/tools/_mock" + "github.com/oxynote/oxynote/server/core/internal/search" "github.com/oxynote/oxynote/server/core/pkg/errutil" "github.com/oxynote/oxynote/server/core/pkg/metricutil" "github.com/oxynote/oxynote/server/core/pkg/testutil" @@ -180,9 +181,12 @@ func testManagerStores() (*Manager, *stores) { log := discardLog() return &Manager{ - log: log, - db: &toolsMock.DB{}, - search: &toolsMock.Searcher{}, + log: log, + db: &toolsMock.DB{}, + search: &toolsMock.Searcher{ + ConfiguredFunc: func() bool { return true }, + }, + jobs: search.NewJobs(true), applier: &toolsMock.EditApplier{}, history: persist.NewHistory(log, st.history), checkpoints: persist.NewCheckpoints(log, st.blobs), @@ -194,11 +198,18 @@ func testManagerStores() (*Manager, *stores) { }, st } +func Test_Manager_Configured(t *testing.T) { + t.Parallel() + + assert.True(t, testManager().Configured()) + assert.False(t, (&Manager{}).Configured()) +} + func Test_NewManager(t *testing.T) { t.Parallel() fc := metricutil.NewFactory("test", prometheus.NewRegistry()) - m := NewManager(discardLog(), nil, &redis.Pool{}, nil, nil, fc, nil, nil, nil, "claude") + m := NewManager(discardLog(), nil, &redis.Pool{}, nil, nil, fc, nil, nil, search.NewJobs(false), nil, "claude") require.NotNil(t, m) assert.NotNil(t, m.log) diff --git a/server/core/internal/assistant/tools/0mock_db_test.go b/server/core/internal/assistant/tools/0mock_db_test.go index 5bf8b217..396ab8bd 100644 --- a/server/core/internal/assistant/tools/0mock_db_test.go +++ b/server/core/internal/assistant/tools/0mock_db_test.go @@ -32,9 +32,6 @@ var _ DB = &DBMock{} // CheckDocumentExistsFunc: func(ctx context.Context, id xid.ID, organizationID string) error { // panic("mock out the CheckDocumentExists method") // }, -// DeleteDocumentFunc: func(ctx context.Context, id xid.ID, organizationID string) error { -// panic("mock out the DeleteDocument method") -// }, // FetchDocumentFunc: func(ctx context.Context, id xid.ID, organizationID string, branchName string) (*document.Document, error) { // panic("mock out the FetchDocument method") // }, @@ -66,9 +63,6 @@ type DBMock struct { // CheckDocumentExistsFunc mocks the CheckDocumentExists method. CheckDocumentExistsFunc func(ctx context.Context, id xid.ID, organizationID string) error - // DeleteDocumentFunc mocks the DeleteDocument method. - DeleteDocumentFunc func(ctx context.Context, id xid.ID, organizationID string) error - // FetchDocumentFunc mocks the FetchDocument method. FetchDocumentFunc func(ctx context.Context, id xid.ID, organizationID string, branchName string) (*document.Document, error) @@ -119,15 +113,6 @@ type DBMock struct { // OrganizationID is the organizationID argument value. OrganizationID string } - // DeleteDocument holds details about calls to the DeleteDocument method. - DeleteDocument []struct { - // Ctx is the ctx argument value. - Ctx context.Context - // ID is the id argument value. - ID xid.ID - // OrganizationID is the organizationID argument value. - OrganizationID string - } // FetchDocument holds details about calls to the FetchDocument method. FetchDocument []struct { // Ctx is the ctx argument value. @@ -195,7 +180,6 @@ type DBMock struct { lockBeginTx sync.RWMutex lockCheckDocumentCycle sync.RWMutex lockCheckDocumentExists sync.RWMutex - lockDeleteDocument sync.RWMutex lockFetchDocument sync.RWMutex lockFetchDocumentTree sync.RWMutex lockFetchDocumentTreeByDocumentParentID sync.RWMutex @@ -335,49 +319,6 @@ func (mock *DBMock) CheckDocumentExistsCalls() []struct { return calls } -// DeleteDocument calls DeleteDocumentFunc. -func (mock *DBMock) DeleteDocument(ctx context.Context, id xid.ID, organizationID string) error { - callInfo := struct { - Ctx context.Context - ID xid.ID - OrganizationID string - }{ - Ctx: ctx, - ID: id, - OrganizationID: organizationID, - } - mock.lockDeleteDocument.Lock() - mock.calls.DeleteDocument = append(mock.calls.DeleteDocument, callInfo) - mock.lockDeleteDocument.Unlock() - if mock.DeleteDocumentFunc == nil { - var ( - errOut error - ) - return errOut - } - return mock.DeleteDocumentFunc(ctx, id, organizationID) -} - -// DeleteDocumentCalls gets all the calls that were made to DeleteDocument. -// Check the length with: -// -// len(mockedDB.DeleteDocumentCalls()) -func (mock *DBMock) DeleteDocumentCalls() []struct { - Ctx context.Context - ID xid.ID - OrganizationID string -} { - var calls []struct { - Ctx context.Context - ID xid.ID - OrganizationID string - } - mock.lockDeleteDocument.RLock() - calls = mock.calls.DeleteDocument - mock.lockDeleteDocument.RUnlock() - return calls -} - // FetchDocument calls FetchDocumentFunc. func (mock *DBMock) FetchDocument(ctx context.Context, id xid.ID, organizationID string, branchName string) (*document.Document, error) { callInfo := struct { diff --git a/server/core/internal/assistant/tools/0mock_searcher_test.go b/server/core/internal/assistant/tools/0mock_searcher_test.go index 19ef5c45..9e62efdd 100644 --- a/server/core/internal/assistant/tools/0mock_searcher_test.go +++ b/server/core/internal/assistant/tools/0mock_searcher_test.go @@ -30,11 +30,17 @@ var _ Searcher = &SearcherMock{} // // } type SearcherMock struct { + // ConfiguredFunc mocks the Configured method. + ConfiguredFunc func() bool + // SearchDocumentBlocksFunc mocks the SearchDocumentBlocks method. SearchDocumentBlocksFunc func(ctx context.Context, organizationID string, query string, limit int) ([]search.Block, error) // calls tracks calls to the methods. calls struct { + // Configured holds details about calls to the Configured method. + Configured []struct { + } // SearchDocumentBlocks holds details about calls to the SearchDocumentBlocks method. SearchDocumentBlocks []struct { // Ctx is the ctx argument value. @@ -47,6 +53,7 @@ type SearcherMock struct { Limit int } } + lockConfigured sync.RWMutex lockSearchDocumentBlocks sync.RWMutex } @@ -97,3 +104,33 @@ func (mock *SearcherMock) SearchDocumentBlocksCalls() []struct { mock.lockSearchDocumentBlocks.RUnlock() return calls } + +// Configured calls ConfiguredFunc. +func (mock *SearcherMock) Configured() bool { + callInfo := struct { + }{} + mock.lockConfigured.Lock() + mock.calls.Configured = append(mock.calls.Configured, callInfo) + mock.lockConfigured.Unlock() + if mock.ConfiguredFunc == nil { + var ( + bOut bool + ) + return bOut + } + return mock.ConfiguredFunc() +} + +// ConfiguredCalls gets all the calls that were made to Configured. +// Check the length with: +// +// len(mockedSearcher.ConfiguredCalls()) +func (mock *SearcherMock) ConfiguredCalls() []struct { +} { + var calls []struct { + } + mock.lockConfigured.RLock() + calls = mock.calls.Configured + mock.lockConfigured.RUnlock() + return calls +} diff --git a/server/core/internal/assistant/tools/0mock_tx_test.go b/server/core/internal/assistant/tools/0mock_tx_test.go index 32c1f821..06621da9 100644 --- a/server/core/internal/assistant/tools/0mock_tx_test.go +++ b/server/core/internal/assistant/tools/0mock_tx_test.go @@ -47,6 +47,9 @@ type TxMock struct { // CommitFunc mocks the Commit method. CommitFunc func() error + // DeleteDocumentFunc mocks the DeleteDocument method. + DeleteDocumentFunc func(ctx context.Context, id xid.ID, organizationID string) ([]xid.ID, error) + // InsertDocumentFunc mocks the InsertDocument method. InsertDocumentFunc func(ctx context.Context, doc document.Document) error @@ -64,6 +67,15 @@ type TxMock struct { // Commit holds details about calls to the Commit method. Commit []struct { } + // DeleteDocument holds details about calls to the DeleteDocument method. + DeleteDocument []struct { + // Ctx is the ctx argument value. + Ctx context.Context + // ID is the id argument value. + ID xid.ID + // OrganizationID is the organizationID argument value. + OrganizationID string + } // InsertDocument holds details about calls to the InsertDocument method. InsertDocument []struct { // Ctx is the ctx argument value. @@ -94,6 +106,7 @@ type TxMock struct { } } lockCommit sync.RWMutex + lockDeleteDocument sync.RWMutex lockInsertDocument sync.RWMutex lockInsertDocumentSearchJob sync.RWMutex lockRollback sync.RWMutex @@ -284,3 +297,47 @@ func (mock *TxMock) UpsertDocumentMaintainersCalls() []struct { mock.lockUpsertDocumentMaintainers.RUnlock() return calls } + +// DeleteDocument calls DeleteDocumentFunc. +func (mock *TxMock) DeleteDocument(ctx context.Context, id xid.ID, organizationID string) ([]xid.ID, error) { + callInfo := struct { + Ctx context.Context + ID xid.ID + OrganizationID string + }{ + Ctx: ctx, + ID: id, + OrganizationID: organizationID, + } + mock.lockDeleteDocument.Lock() + mock.calls.DeleteDocument = append(mock.calls.DeleteDocument, callInfo) + mock.lockDeleteDocument.Unlock() + if mock.DeleteDocumentFunc == nil { + var ( + iDsOut []xid.ID + errOut error + ) + return iDsOut, errOut + } + return mock.DeleteDocumentFunc(ctx, id, organizationID) +} + +// DeleteDocumentCalls gets all the calls that were made to DeleteDocument. +// Check the length with: +// +// len(mockedTx.DeleteDocumentCalls()) +func (mock *TxMock) DeleteDocumentCalls() []struct { + Ctx context.Context + ID xid.ID + OrganizationID string +} { + var calls []struct { + Ctx context.Context + ID xid.ID + OrganizationID string + } + mock.lockDeleteDocument.RLock() + calls = mock.calls.DeleteDocument + mock.lockDeleteDocument.RUnlock() + return calls +} diff --git a/server/core/internal/assistant/tools/_mock/db.go b/server/core/internal/assistant/tools/_mock/db.go index d823347a..b6750adb 100644 --- a/server/core/internal/assistant/tools/_mock/db.go +++ b/server/core/internal/assistant/tools/_mock/db.go @@ -33,9 +33,6 @@ var _ tools.DB = &DB{} // CheckDocumentExistsFunc: func(ctx context.Context, id xid.ID, organizationID string) error { // panic("mock out the CheckDocumentExists method") // }, -// DeleteDocumentFunc: func(ctx context.Context, id xid.ID, organizationID string) error { -// panic("mock out the DeleteDocument method") -// }, // FetchDocumentFunc: func(ctx context.Context, id xid.ID, organizationID string, branchName string) (*document.Document, error) { // panic("mock out the FetchDocument method") // }, @@ -67,9 +64,6 @@ type DB struct { // CheckDocumentExistsFunc mocks the CheckDocumentExists method. CheckDocumentExistsFunc func(ctx context.Context, id xid.ID, organizationID string) error - // DeleteDocumentFunc mocks the DeleteDocument method. - DeleteDocumentFunc func(ctx context.Context, id xid.ID, organizationID string) error - // FetchDocumentFunc mocks the FetchDocument method. FetchDocumentFunc func(ctx context.Context, id xid.ID, organizationID string, branchName string) (*document.Document, error) @@ -120,15 +114,6 @@ type DB struct { // OrganizationID is the organizationID argument value. OrganizationID string } - // DeleteDocument holds details about calls to the DeleteDocument method. - DeleteDocument []struct { - // Ctx is the ctx argument value. - Ctx context.Context - // ID is the id argument value. - ID xid.ID - // OrganizationID is the organizationID argument value. - OrganizationID string - } // FetchDocument holds details about calls to the FetchDocument method. FetchDocument []struct { // Ctx is the ctx argument value. @@ -196,7 +181,6 @@ type DB struct { lockBeginTx sync.RWMutex lockCheckDocumentCycle sync.RWMutex lockCheckDocumentExists sync.RWMutex - lockDeleteDocument sync.RWMutex lockFetchDocument sync.RWMutex lockFetchDocumentTree sync.RWMutex lockFetchDocumentTreeByDocumentParentID sync.RWMutex @@ -336,49 +320,6 @@ func (mock *DB) CheckDocumentExistsCalls() []struct { return calls } -// DeleteDocument calls DeleteDocumentFunc. -func (mock *DB) DeleteDocument(ctx context.Context, id xid.ID, organizationID string) error { - callInfo := struct { - Ctx context.Context - ID xid.ID - OrganizationID string - }{ - Ctx: ctx, - ID: id, - OrganizationID: organizationID, - } - mock.lockDeleteDocument.Lock() - mock.calls.DeleteDocument = append(mock.calls.DeleteDocument, callInfo) - mock.lockDeleteDocument.Unlock() - if mock.DeleteDocumentFunc == nil { - var ( - errOut error - ) - return errOut - } - return mock.DeleteDocumentFunc(ctx, id, organizationID) -} - -// DeleteDocumentCalls gets all the calls that were made to DeleteDocument. -// Check the length with: -// -// len(mockedDB.DeleteDocumentCalls()) -func (mock *DB) DeleteDocumentCalls() []struct { - Ctx context.Context - ID xid.ID - OrganizationID string -} { - var calls []struct { - Ctx context.Context - ID xid.ID - OrganizationID string - } - mock.lockDeleteDocument.RLock() - calls = mock.calls.DeleteDocument - mock.lockDeleteDocument.RUnlock() - return calls -} - // FetchDocument calls FetchDocumentFunc. func (mock *DB) FetchDocument(ctx context.Context, id xid.ID, organizationID string, branchName string) (*document.Document, error) { callInfo := struct { diff --git a/server/core/internal/assistant/tools/_mock/searcher.go b/server/core/internal/assistant/tools/_mock/searcher.go index 495421c1..0c8385a5 100644 --- a/server/core/internal/assistant/tools/_mock/searcher.go +++ b/server/core/internal/assistant/tools/_mock/searcher.go @@ -31,11 +31,17 @@ var _ tools.Searcher = &Searcher{} // // } type Searcher struct { + // ConfiguredFunc mocks the Configured method. + ConfiguredFunc func() bool + // SearchDocumentBlocksFunc mocks the SearchDocumentBlocks method. SearchDocumentBlocksFunc func(ctx context.Context, organizationID string, query string, limit int) ([]search.Block, error) // calls tracks calls to the methods. calls struct { + // Configured holds details about calls to the Configured method. + Configured []struct { + } // SearchDocumentBlocks holds details about calls to the SearchDocumentBlocks method. SearchDocumentBlocks []struct { // Ctx is the ctx argument value. @@ -48,6 +54,7 @@ type Searcher struct { Limit int } } + lockConfigured sync.RWMutex lockSearchDocumentBlocks sync.RWMutex } @@ -98,3 +105,33 @@ func (mock *Searcher) SearchDocumentBlocksCalls() []struct { mock.lockSearchDocumentBlocks.RUnlock() return calls } + +// Configured calls ConfiguredFunc. +func (mock *Searcher) Configured() bool { + callInfo := struct { + }{} + mock.lockConfigured.Lock() + mock.calls.Configured = append(mock.calls.Configured, callInfo) + mock.lockConfigured.Unlock() + if mock.ConfiguredFunc == nil { + var ( + bOut bool + ) + return bOut + } + return mock.ConfiguredFunc() +} + +// ConfiguredCalls gets all the calls that were made to Configured. +// Check the length with: +// +// len(mockedSearcher.ConfiguredCalls()) +func (mock *Searcher) ConfiguredCalls() []struct { +} { + var calls []struct { + } + mock.lockConfigured.RLock() + calls = mock.calls.Configured + mock.lockConfigured.RUnlock() + return calls +} diff --git a/server/core/internal/assistant/tools/document_test.go b/server/core/internal/assistant/tools/document_test.go index f0106658..c8182cde 100644 --- a/server/core/internal/assistant/tools/document_test.go +++ b/server/core/internal/assistant/tools/document_test.go @@ -464,6 +464,30 @@ func Test_deleteDocument_Execute(t *testing.T) { docID := xid.New() + // stubDeleteDB wires the given DB (nil for a bare one) with a + // transaction whose delete reports one destroyed document, or fails + // with deleteErr. + stubDeleteDB := func(db *DBMock, deleteErr error) *DBMock { + if db == nil { + db = &DBMock{} + } + + db.BeginTxFunc = func(_ context.Context, dest any) error { + reflect.ValueOf(dest).Elem().Set(reflect.ValueOf(&TxMock{ + DeleteDocumentFunc: func(_ context.Context, id xid.ID, _ string) ([]xid.ID, error) { + return []xid.ID{id}, deleteErr + }, + InsertDocumentSearchJobFunc: func(context.Context, search.BlocksDifference) error { + return nil + }, + })) + + return nil + } + + return db + } + cc := map[string]struct { DB *DBMock Args string @@ -476,26 +500,22 @@ func Test_deleteDocument_Execute(t *testing.T) { Args: `{"document_id":"nope"}`, Err: assert.AnError, }, - "Error returned by db.DeleteDocument": { - DB: &DBMock{ - DeleteDocumentFunc: func(context.Context, xid.ID, string) error { - return assert.AnError - }, - }, + "Error returned by Tx.DeleteDocument": { + DB: stubDeleteDB(nil, assert.AnError), Args: `{"document_id":"` + docID.String() + `"}`, Err: assert.AnError, }, "Deleted": { - DB: stubDocumentDB(), + DB: stubDeleteDB(stubDocumentDB(), nil), Args: `{"document_id":"` + docID.String() + `"}`, Notify: 1, }, "Deleted even when the parent cannot be captured": { - DB: &DBMock{ + DB: stubDeleteDB(&DBMock{ FetchDocumentFunc: func(context.Context, xid.ID, string, string) (*document.Document, error) { return nil, assert.AnError }, - }, + }, nil), Args: `{"document_id":"` + docID.String() + `"}`, Notify: 1, }, diff --git a/server/core/internal/assistant/tools/input.go b/server/core/internal/assistant/tools/input.go index c4585210..da39cdb2 100644 --- a/server/core/internal/assistant/tools/input.go +++ b/server/core/internal/assistant/tools/input.go @@ -39,6 +39,11 @@ type Deps struct { // search is the full-text index behind search_documents. search Searcher + // jobs is the way into the search-job queue: it decides whether a + // deployment indexes at all, so writes queue through it instead of + // hitting the database directly. + jobs *search.Jobs + // runners hands out the runner a data-source tool reads through. runners DataSourceRunners @@ -70,6 +75,7 @@ func NewDeps( log *slog.Logger, db DB, searcher Searcher, + jobs *search.Jobs, runners DataSourceRunners, applier EditApplier, tree TreeNotifier, @@ -84,6 +90,7 @@ func NewDeps( ), db: db, search: searcher, + jobs: jobs, runners: runners, applier: applier, tree: tree, @@ -281,7 +288,7 @@ func (i *input) CreateDocument(doc document.Document) error { // without this the document is invisible to search until someone // edits it, since only the persist path queues a job. - if err := tx.InsertDocumentSearchJob(i.ctx, search.BlocksDiff(nil, doc.Search())); err != nil { + if err := i.jobs.Enqueue(i.ctx, tx, search.BlocksDiff(nil, doc.Search())); err != nil { return fmt.Errorf("insert search job: %w", err) } @@ -296,8 +303,32 @@ func (i *input) CreateDocument(doc document.Document) error { // DeleteDocument removes the document. It records nothing as touched: // the document is gone, so there is nothing left to point a caller at. +// The delete reports the ids of the destroyed subtree, and their +// search-index removal is queued in the same transaction: after the +// commit nothing else knows what went away. func (i *input) DeleteDocument(id xid.ID) error { - return i.db.DeleteDocument(i.ctx, id, i.orgID) + var tx Tx + + if err := i.db.BeginTx(i.ctx, &tx); err != nil { + return err + } + + defer tx.Rollback() //nolint:errcheck // error provides no meaningful info + + ids, err := tx.DeleteDocument(i.ctx, id, i.orgID) + if err != nil { + return err + } + + if len(ids) != 0 { + if err := i.jobs.Enqueue(i.ctx, tx, search.BlocksDifference{ + RemovedDocuments: ids, + }); err != nil { + return fmt.Errorf("insert search job: %w", err) + } + } + + return tx.Commit() } // MoveDocument re-parents the document. @@ -495,7 +526,6 @@ type DataSourceRunners interface { // agent satisfies it. // //go:generate ../../../scripts/codegen/mock -t both DB db -//nolint:interfacebloat // one persistence surface per package; splitting it by domain would only restate the same list under two names type DB interface { sqlutil.DB @@ -519,9 +549,6 @@ type DB interface { // content tree (no branch metadata). FetchMainBranchContent(ctx context.Context, docID xid.ID, organizationID string) (document.Content, error) - // DeleteDocument should remove a document. Used by delete_document. - DeleteDocument(ctx context.Context, id xid.ID, organizationID string) error - // UpdateDocumentParentID should re-parent a document. Used by // move_document. UpdateDocumentParentID(ctx context.Context, id xid.ID, parentID null.Value[xid.ID], organizationID string) error @@ -560,8 +587,13 @@ type Tx interface { UpsertDocumentMaintainers(ctx context.Context, documentID xid.ID, organizationID string, maintainerIDs []string) error // InsertDocumentSearchJob should queue the search index update for a - // document. Used by create_document. + // document. Used by create_document and delete_document. InsertDocumentSearchJob(ctx context.Context, diff search.BlocksDifference) error + + // DeleteDocument should remove a document and report the ids of the + // document and of every cascade-deleted descendant. Used by + // delete_document. + DeleteDocument(ctx context.Context, id xid.ID, organizationID string) ([]xid.ID, error) } // Searcher is the full-text search surface search_documents uses. @@ -569,6 +601,10 @@ type Tx interface { // //go:generate ../../../scripts/codegen/mock -t both Searcher searcher type Searcher interface { + // Configured should report whether search is configured on this + // deployment. + Configured() bool + // SearchDocumentBlocks should return blocks whose text matches the // query, scoped to the organization and capped at limit hits. SearchDocumentBlocks(ctx context.Context, organizationID, query string, limit int) ([]search.Block, error) diff --git a/server/core/internal/assistant/tools/input_test.go b/server/core/internal/assistant/tools/input_test.go index f28d4a01..4ed3e442 100644 --- a/server/core/internal/assistant/tools/input_test.go +++ b/server/core/internal/assistant/tools/input_test.go @@ -47,9 +47,12 @@ func testDeps(db *DBMock, applier *EditApplierMock, tree *TreeNotifierMock) *Dep } d := &Deps{ - log: discardLog(), - db: db, - search: &SearcherMock{}, + log: discardLog(), + db: db, + search: &SearcherMock{ + ConfiguredFunc: func() bool { return true }, + }, + jobs: search.NewJobs(true), runners: &DataSourceRunnersMock{ RunnerFunc: func(datasource.DataSource) datasource.Runner { return &datasourceMock.Runner{} @@ -205,12 +208,15 @@ func Test_NewDeps(t *testing.T) { RunnerFunc: func(datasource.DataSource) datasource.Runner { return runner }, } - d := NewDeps(discardLog(), db, searcher, runners, applier, tree, offload, "org", "user") + jobs := search.NewJobs(true) + + d := NewDeps(discardLog(), db, searcher, jobs, runners, applier, tree, offload, "org", "user") require.NotNil(t, d) assert.NotNil(t, d.log) assert.Same(t, db, d.db) assert.Same(t, searcher, d.search) + assert.Same(t, jobs, d.jobs) assert.Same(t, runners, d.runners) assert.Same(t, applier, d.applier) assert.Same(t, tree, d.tree) @@ -404,29 +410,102 @@ func Test_input_DeleteDocument(t *testing.T) { docID := xid.New() - // error - inp := testInput(testDeps(&DBMock{ - DeleteDocumentFunc: func(context.Context, xid.ID, string) error { - return assert.AnError - }, - }, nil, nil), NameDeleteDocument, `{}`) - - testutil.AssertEqualError(t, assert.AnError, inp.DeleteDocument(docID)) - - // success: the document is gone, so nothing is recorded as touched - // — a link to it would only point at something that no longer - // exists. - db := &DBMock{ - DeleteDocumentFunc: func(context.Context, xid.ID, string) error { - return nil - }, + type tcase struct { + DB *DBMock + Tx *TxMock + Commits int + Jobs int + Err error } - inp = testInput(testDeps(db, nil, nil), NameDeleteDocument, `{}`) + stubTx := func(ids []xid.ID, deleteErr, jobErr, commitErr error) *TxMock { + return &TxMock{ + DeleteDocumentFunc: func(context.Context, xid.ID, string) ([]xid.ID, error) { + return ids, deleteErr + }, + InsertDocumentSearchJobFunc: func(context.Context, search.BlocksDifference) error { + return jobErr + }, + CommitFunc: func() error { return commitErr }, + } + } - require.NoError(t, inp.DeleteDocument(docID)) - assert.Len(t, db.DeleteDocumentCalls(), 1) - assert.Empty(t, inp.touched) + stubDB := func(tx *TxMock, beginErr error) *DBMock { + return &DBMock{ + BeginTxFunc: func(_ context.Context, dest any) error { + if beginErr != nil { + return beginErr + } + + reflect.ValueOf(dest).Elem().Set(reflect.ValueOf(tx)) + + return nil + }, + } + } + + cc := map[string]tcase{ + "Error returned by db.BeginTx": func() tcase { + return tcase{DB: stubDB(nil, assert.AnError), Err: assert.AnError} + }(), + "Error returned by Tx.DeleteDocument": func() tcase { + tx := stubTx(nil, assert.AnError, nil, nil) + + return tcase{DB: stubDB(tx, nil), Tx: tx, Err: assert.AnError} + }(), + "Error returned by Tx.InsertDocumentSearchJob": func() tcase { + tx := stubTx([]xid.ID{docID}, nil, assert.AnError, nil) + + return tcase{DB: stubDB(tx, nil), Tx: tx, Jobs: 1, Err: assert.AnError} + }(), + "Error returned by Tx.Commit": func() tcase { + tx := stubTx([]xid.ID{docID}, nil, nil, assert.AnError) + + return tcase{DB: stubDB(tx, nil), Tx: tx, Commits: 1, Jobs: 1, Err: assert.AnError} + }(), + // a document unknown to the delete reports no subtree, so there + // is no removal to queue. + "Empty subtree queues nothing": func() tcase { + tx := stubTx(nil, nil, nil, nil) + + return tcase{DB: stubDB(tx, nil), Tx: tx, Commits: 1} + }(), + "Delete and search removal land together": func() tcase { + tx := stubTx([]xid.ID{docID, xid.New()}, nil, nil, nil) + + return tcase{DB: stubDB(tx, nil), Tx: tx, Commits: 1, Jobs: 1} + }(), + } + + for cn, c := range cc { + t.Run(cn, func(t *testing.T) { + t.Parallel() + + inp := testInput(testDeps(c.DB, nil, nil), NameDeleteDocument, `{}`) + + err := inp.DeleteDocument(docID) + testutil.AssertEqualError(t, c.Err, err) + + // the document is gone, so nothing is recorded as touched — + // a link to it would only point at something that no longer + // exists. + assert.Empty(t, inp.touched) + + if c.Tx == nil { + return + } + + assert.Len(t, c.Tx.CommitCalls(), c.Commits) + assert.Len(t, c.Tx.RollbackCalls(), 1) + + ff := c.Tx.InsertDocumentSearchJobCalls() + require.Len(t, ff, c.Jobs) + + if c.Jobs != 0 { + assert.NotEmpty(t, ff[0].Diff.RemovedDocuments) + } + }) + } } func Test_input_recordTouched(t *testing.T) { diff --git a/server/core/internal/assistant/tools/tools.go b/server/core/internal/assistant/tools/tools.go index cbba0e0b..a6f57e0a 100644 --- a/server/core/internal/assistant/tools/tools.go +++ b/server/core/internal/assistant/tools/tools.go @@ -99,8 +99,16 @@ func New(deps *Deps) *Set { getDocument{}, readDocumentSummary{}, readBlock{}, - searchDocuments{}, + } + // a deployment without search has no index behind search_documents, + // so the tool is left out of the registry entirely — neither the + // model nor the MCP surface is offered a tool that cannot work. + if deps.search.Configured() { + all = append(all, searchDocuments{}) + } + + all = append(all, listDataSources{}, getPrometheusMetadata{}, listPrometheusLabelNames{}, @@ -125,7 +133,7 @@ func New(deps *Deps) *Set { deleteBlock{}, readToolOutput{}, - } + ) s := &Set{tools: make(map[Name]registryTool, len(all))} diff --git a/server/core/internal/assistant/tools/tools_test.go b/server/core/internal/assistant/tools/tools_test.go index c9b73553..cfd06110 100644 --- a/server/core/internal/assistant/tools/tools_test.go +++ b/server/core/internal/assistant/tools/tools_test.go @@ -112,6 +112,26 @@ func Test_New(t *testing.T) { } } +func Test_New_withoutSearch(t *testing.T) { + t.Parallel() + + d := testDeps(nil, nil, nil) + d.search = &SearcherMock{} + + s := New(d) + + // without search there is no index behind search_documents, so the + // registry offers it to no surface at all; everything else stays. + require.Len(t, s.tools, len(allToolNames())-1) + + _, ok := s.tools[NameSearchDocuments] + assert.False(t, ok) + + for _, e := range s.Entries() { + assert.NotEqual(t, string(NameSearchDocuments), e.Name) + } +} + func Test_Set_Tools(t *testing.T) { t.Parallel() diff --git a/server/core/internal/db/document.go b/server/core/internal/db/document.go index 75e24017..907df21c 100644 --- a/server/core/internal/db/document.go +++ b/server/core/internal/db/document.go @@ -10,7 +10,6 @@ import ( "github.com/jackc/pgx/v5/pgconn" "github.com/jmoiron/sqlx" "github.com/oxynote/oxynote/server/core/internal/document" - "github.com/oxynote/oxynote/server/core/internal/search" "github.com/oxynote/oxynote/server/core/pkg/sqlutil" "github.com/rs/xid" ) @@ -293,13 +292,19 @@ func (a *agent) UpdateDocumentParentID(ctx context.Context, id xid.ID, parentID }) } -// DeleteDocument deletes a document from the database and queues the removal -// of its search index entries. Deleting the row cascades to the descendants, -// so the removal is queued here rather than at the call sites: after the -// delete there is nothing left to tell the index which documents went away. -func (a *agent) DeleteDocument(ctx context.Context, id xid.ID, organizationID string) error { - return sqlutil.WrapTx(ctx, a.sql, func(tx *sqlx.Tx) error { - ids, err := a.fetchDocumentSubtreeIDs(ctx, tx, id, organizationID) +// DeleteDocument deletes a document from the database, returning the ids +// of the document and of every descendant the delete cascaded to. +// Deleting the row destroys the subtree, so the ids are collected here +// rather than by the caller: after the delete there is nothing left to +// tell the search index which documents went away. The caller queues the +// index removal from them, in the same transaction when it runs in one. +func (a *agent) DeleteDocument(ctx context.Context, id xid.ID, organizationID string) ([]xid.ID, error) { + var ids []xid.ID + + err := sqlutil.WrapTx(ctx, a.sql, func(tx *sqlx.Tx) error { + var err error + + ids, err = a.fetchDocumentSubtreeIDs(ctx, tx, id, organizationID) if err != nil { return err } @@ -310,18 +315,15 @@ func (a *agent) DeleteDocument(ctx context.Context, id xid.ID, organizationID st "fk_organization_id": organizationID, }).MustSql() - if _, err = tx.ExecContext(ctx, q, args...); err != nil { - return err - } + _, err = tx.ExecContext(ctx, q, args...) - if len(ids) == 0 { - return nil - } - - return a.insertDocumentSearchJob(ctx, tx, search.BlocksDifference{ - RemovedDocuments: ids, - }) + return err }) + if err != nil { + return nil, err + } + + return ids, nil } // fetchDocumentSubtreeIDs retrieves the IDs of the document and of every diff --git a/server/core/internal/db/document_test.go b/server/core/internal/db/document_test.go index 99e11c80..15a12dfc 100644 --- a/server/core/internal/db/document_test.go +++ b/server/core/internal/db/document_test.go @@ -851,7 +851,7 @@ func Test_agent_DeleteDocument(t *testing.T) { OrganizationID string FileID string HookID xid.ID - SearchRemoved []xid.ID + SubtreeIDs []xid.ID Err error } @@ -862,12 +862,13 @@ func Test_agent_DeleteDocument(t *testing.T) { return tcase{ ID: doc.ID, OrganizationID: doc.OrganizationID, - SearchRemoved: []xid.ID{doc.ID}, + SubtreeIDs: []xid.ID{doc.ID}, } }, - // the descendants go away with the cascade, so their index entries - // have to be queued for removal by the delete that destroys them. - "Descendants are queued for search removal": func(t *testing.T, db *DB) tcase { + // the descendants go away with the cascade, so the delete has to + // report them: the caller queues the search-index removal from + // the returned ids, and nothing else knows what was destroyed. + "Descendants are reported for search removal": func(t *testing.T, db *DB) tcase { organizationID := prepOrganizations(t, db, 1)[0] docs := prepDocuments(t, db, 4, func(_ int, doc *document.Document) { @@ -881,10 +882,10 @@ func Test_agent_DeleteDocument(t *testing.T) { return tcase{ ID: docs[0].ID, OrganizationID: organizationID, - SearchRemoved: []xid.ID{docs[0].ID, docs[1].ID, docs[2].ID}, + SubtreeIDs: []xid.ID{docs[0].ID, docs[1].ID, docs[2].ID}, } }, - "Unknown document queues nothing": func(t *testing.T, db *DB) tcase { + "Unknown document reports nothing": func(t *testing.T, db *DB) tcase { doc := prepDocuments(t, db, 1, nil)[0] return tcase{ @@ -914,7 +915,7 @@ func Test_agent_DeleteDocument(t *testing.T) { OrganizationID: doc.OrganizationID, FileID: f.ID, HookID: hk.ID, - SearchRemoved: []xid.ID{doc.ID}, + SubtreeIDs: []xid.ID{doc.ID}, } }, } @@ -926,13 +927,15 @@ func Test_agent_DeleteDocument(t *testing.T) { db := prepTempDB(t) c := cfn(t, db) - err := db.DeleteDocument(context.Background(), c.ID, c.OrganizationID) + ids, err := db.DeleteDocument(context.Background(), c.ID, c.OrganizationID) testutil.RequireEqualError(t, c.Err, err) if err != nil { return } + assert.ElementsMatch(t, c.SubtreeIDs, ids) + var doc document.Document q, args := db.selectDocumentWithBranch(db.builder.Select(), document.DefaultBranch). @@ -944,15 +947,11 @@ func Test_agent_DeleteDocument(t *testing.T) { err = db.sql.Get(&doc, q, args...) require.Equal(t, sql.ErrNoRows, err) + // the delete itself queues nothing: the caller owns the + // search-index removal. jobs, err := db.FetchDocumentSearchJobs(context.Background(), 0, 10) require.NoError(t, err) - - if len(c.SearchRemoved) == 0 { - assert.Empty(t, jobs) - } else { - require.Len(t, jobs, 1) - assert.ElementsMatch(t, c.SearchRemoved, jobs[0].BlockDiff.RemovedDocuments) - } + assert.Empty(t, jobs) if c.FileID == "" { return diff --git a/server/core/internal/document/hook/manager/manager.go b/server/core/internal/document/hook/manager/manager.go index 8623e75d..52b4f312 100644 --- a/server/core/internal/document/hook/manager/manager.go +++ b/server/core/internal/document/hook/manager/manager.go @@ -110,13 +110,7 @@ func (m *Manager) processHooks(ctx context.Context) error { continue } - // github-tracking hooks cannot make progress without a - // configured GitHub App; skip them and leave their state - // untouched. - if h.Type == hook.TypeGithubTracking && !m.githubMan.Configured() { - m.log.With("hook_id", h.ID). - Warn("skipping github-tracking hook: github app is not configured") - + if m.skipsUnconfigured(&h) { continue } @@ -246,6 +240,26 @@ func (m *Manager) ensureHook( // row describing it. The external teardown goes first: the row is the only // record of the resource, so dropping it first would strand the watcher // with nothing left to find it by. +// skipsUnconfigured reports whether the hook depends on an integration +// this deployment does not have. Such a hook cannot make progress, so +// the processing pass skips it and leaves its state untouched. +func (m *Manager) skipsUnconfigured(h *hook.Hook) bool { + switch { + case h.Type == hook.TypeGithubTracking && !m.githubMan.Configured(): + m.log.With("hook_id", h.ID). + Warn("skipping github-tracking hook: github app is not configured") + + return true + case h.Type == hook.TypeURLWatcher && !m.webchangeClient.Configured(): + m.log.With("hook_id", h.ID). + Warn("skipping url-watcher hook: changedetection is not configured") + + return true + default: + return false + } +} + func (m *Manager) deleteHook(ctx context.Context, h *hook.Hook) { err := h.Delete(ctx, hook.NewInput( h.OrganizationID.String, diff --git a/server/core/internal/document/hook/manager/manager_test.go b/server/core/internal/document/hook/manager/manager_test.go index c979873b..2a5f4368 100644 --- a/server/core/internal/document/hook/manager/manager_test.go +++ b/server/core/internal/document/hook/manager/manager_test.go @@ -10,6 +10,7 @@ import ( "github.com/guregu/null/v5" "github.com/oxynote/oxynote/server/core/internal/apps/github" + "github.com/oxynote/oxynote/server/core/internal/apps/webchange" "github.com/oxynote/oxynote/server/core/internal/document" "github.com/oxynote/oxynote/server/core/internal/document/hook" "github.com/oxynote/oxynote/server/core/internal/document/hook/processor" @@ -72,15 +73,30 @@ func stubDocument() *document.Document { } } -// newTestManager creates a Manager with an unconfigured GitHub manager and -// the given mocks. +// newTestManager creates a Manager with unconfigured GitHub and +// changedetection clients and the given mocks. func newTestManager(t *testing.T, db *DBMock, pub *fakePublisher) *Manager { t.Helper() githubMan, err := github.NewManager(nil, github.Options{}) require.NoError(t, err) - return NewManager(slog.New(slog.DiscardHandler), db, githubMan, nil, pub) + return NewManager(slog.New(slog.DiscardHandler), db, githubMan, webchange.NewClient("", ""), pub) +} + +// urlWatcherHook builds a url-watcher hook that already holds a +// changedetection.io watcher in its state. +func urlWatcherHook(branchID xid.ID) hook.Hook { + return hook.Hook{ + ID: xid.New(), + Type: hook.TypeURLWatcher, + DocumentID: null.ValueFrom(xid.New()), + OrganizationID: null.StringFrom("org-1"), + BranchID: null.ValueFrom(branchID), + Settings: processor.Settings(`{"url":"https://example.com"}`), + State: processor.State(`{"watcherId":"w1"}`), + Score: mathutil.Hundred, + } } func Test_Manager_Start(t *testing.T) { @@ -267,6 +283,33 @@ func Test_Manager_processHooks(t *testing.T) { wasUpdateCalled(0), ), }, + "URL watcher hook is skipped when unconfigured": { + Hooks: func(_ *testing.T) []hook.Hook { + return []hook.Hook{urlWatcherHook(branchID)} + }, + Checks: checks( + hasError(false), + wasFetchDocumentCalled(0), + wasUpdateCalled(0), + ), + }, + + // an unconfigured changedetection must not trap the orphaned row: + // the teardown is skipped and the row is still deleted. + "Orphaned URL watcher is deleted when unconfigured": { + Hooks: func(_ *testing.T) []hook.Hook { + h := urlWatcherHook(branchID) + h.BranchID = null.Value[xid.ID]{} + + return []hook.Hook{h} + }, + Checks: checks( + hasError(false), + wasDeleteCalled(1), + wasFetchDocumentCalled(0), + wasUpdateCalled(0), + ), + }, "Missing document deletes the hook": { Hooks: func(t *testing.T) []hook.Hook { return []hook.Hook{stubHook(t, branchID, time.Now().Add(time.Hour), time.Now())} diff --git a/server/core/internal/search/client.go b/server/core/internal/search/client.go index 7a56899c..65cff36e 100644 --- a/server/core/internal/search/client.go +++ b/server/core/internal/search/client.go @@ -7,16 +7,22 @@ import ( "encoding/json" "errors" "fmt" + "net/http" "strconv" "strings" "time" "github.com/meilisearch/meilisearch-go" + "github.com/oxynote/oxynote/server/core/pkg/errutil" ) //go:embed static/synonyms.json var synonymsFile []byte +// ErrNotConfigured is returned when search is not configured on this +// deployment. +var ErrNotConfigured = errutil.New(http.StatusConflict, "search.not_configured", "search is not configured") + // _documentsIndex is the name of the documents index. const _documentsIndex = "documents" @@ -42,7 +48,10 @@ type Client struct { meiliMan meilisearch.ServiceManager } -// NewClient creates a new Meilisearch client wrapper. +// NewClient creates a new Meilisearch client wrapper. A nil meiliMan means +// search is not configured on this deployment: the client is returned +// without touching Meilisearch and every call on it reports +// ErrNotConfigured. func NewClient( ctx context.Context, meiliMan meilisearch.ServiceManager, @@ -51,6 +60,10 @@ func NewClient( meiliMan: meiliMan, } + if !c.Configured() { + return c, nil + } + if err := c.ensureIndex(ctx); err != nil { return nil, fmt.Errorf("ensuring documents index: %w", err) } @@ -58,6 +71,11 @@ func NewClient( return c, nil } +// Configured reports whether search is configured on this deployment. +func (c *Client) Configured() bool { + return c.meiliMan != nil +} + // ensureIndex ensures the documents index exists. func (c *Client) ensureIndex(ctx context.Context) error { _, err := c.meiliMan.GetIndexWithContext(ctx, _documentsIndex) @@ -143,6 +161,10 @@ func (c *Client) setupSynonyms(ctx context.Context, index string) error { // SearchDocuments searches documents in the index. func (c *Client) SearchDocuments(ctx context.Context, organizationID, query string) ([]byte, error) { + if !c.Configured() { + return nil, ErrNotConfigured + } + res, err := c.meiliMan.Index(_documentsIndex).SearchWithContext(ctx, query, &meilisearch.SearchRequest{ AttributesToSearchOn: []string{_textAttribute}, AttributesToCrop: []string{_textAttribute}, @@ -180,6 +202,10 @@ func (c *Client) SearchDocuments(ctx context.Context, organizationID, query stri // callers — the AI assistant's search tool — can consume the text // directly. func (c *Client) SearchDocumentBlocks(ctx context.Context, organizationID, query string, limit int) ([]Block, error) { + if !c.Configured() { + return nil, ErrNotConfigured + } + res, err := c.meiliMan.Index(_documentsIndex).SearchWithContext(ctx, query, &meilisearch.SearchRequest{ AttributesToSearchOn: []string{_textAttribute}, Filter: fmt.Sprintf("organizationId = %q", organizationID), @@ -207,6 +233,10 @@ func (c *Client) SearchDocumentBlocks(ctx context.Context, organizationID, query // ReplaceDocumentBlocks replaces document blocks in the index based on the // provided differences. func (c *Client) ReplaceDocumentBlocks(ctx context.Context, bd BlocksDifference) error { + if !c.Configured() { + return ErrNotConfigured + } + if len(bd.Added) != 0 { _, err := c.meiliMan.Index(_documentsIndex).AddDocumentsWithContext(ctx, bd.Added, nil) if err != nil { diff --git a/server/core/internal/search/client_test.go b/server/core/internal/search/client_test.go index 3e53bdea..33715661 100644 --- a/server/core/internal/search/client_test.go +++ b/server/core/internal/search/client_test.go @@ -36,6 +36,27 @@ func stubManagers(idx *mock.MeiliIndexManager) *mock.MeiliServiceManager { func Test_NewClient(t *testing.T) { t.Parallel() + // a nil service manager is the disabled mode: nothing is set up and + // every call refuses with the sentinel. + t.Run("Unconfigured client", func(t *testing.T) { + t.Parallel() + + c, err := NewClient(context.Background(), nil) + require.NoError(t, err) + require.NotNil(t, c) + assert.False(t, c.Configured()) + + data, serr := c.SearchDocuments(context.Background(), "org1", "query") + assert.ErrorIs(t, serr, ErrNotConfigured) + assert.Nil(t, data) + + blocks, serr := c.SearchDocumentBlocks(context.Background(), "org1", "query", 5) + assert.ErrorIs(t, serr, ErrNotConfigured) + assert.Nil(t, blocks) + + assert.ErrorIs(t, c.ReplaceDocumentBlocks(context.Background(), BlocksDifference{}), ErrNotConfigured) + }) + type check func(*testing.T, *mock.MeiliServiceManager, *mock.MeiliIndexManager, error) checks := func(cc ...check) []check { return cc } diff --git a/server/core/internal/search/jobs.go b/server/core/internal/search/jobs.go new file mode 100644 index 00000000..4bb7bf1a --- /dev/null +++ b/server/core/internal/search/jobs.go @@ -0,0 +1,38 @@ +package search + +import "context" + +// Jobs is the way into the document search-job queue: every enqueue in +// the codebase goes through it, so whether a deployment indexes at all +// is decided here rather than by each call site or the database layer. +type Jobs struct { + enabled bool +} + +// NewJobs creates a fresh instance of Jobs. When enabled is false — +// search is not configured on this deployment — every enqueued job is +// dropped instead of piling up in a table nothing consumes; documents +// created meanwhile were never indexed, so there is nothing stale to +// clean up either. +func NewJobs(enabled bool) *Jobs { + return &Jobs{ + enabled: enabled, + } +} + +// Enqueue queues the diff through the given inserter, which is the +// caller's own database handle so the job commits atomically with the +// content change it describes. +func (j *Jobs) Enqueue(ctx context.Context, ins JobInserter, diff BlocksDifference) error { + if !j.enabled { + return nil + } + + return ins.InsertDocumentSearchJob(ctx, diff) +} + +// JobInserter is the database surface Enqueue writes through. +type JobInserter interface { + // InsertDocumentSearchJob should insert the document search job. + InsertDocumentSearchJob(ctx context.Context, diff BlocksDifference) error +} diff --git a/server/core/internal/search/jobs_test.go b/server/core/internal/search/jobs_test.go new file mode 100644 index 00000000..997ee2fd --- /dev/null +++ b/server/core/internal/search/jobs_test.go @@ -0,0 +1,49 @@ +package search + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// insertRecorder records InsertDocumentSearchJob calls for Jobs tests. +type insertRecorder struct { + diffs []BlocksDifference + err error +} + +func (ir *insertRecorder) InsertDocumentSearchJob(_ context.Context, diff BlocksDifference) error { + ir.diffs = append(ir.diffs, diff) + + return ir.err +} + +func Test_NewJobs(t *testing.T) { + t.Parallel() + + j := NewJobs(true) + require.NotNil(t, j) + assert.True(t, j.enabled) +} + +func Test_Jobs_Enqueue(t *testing.T) { + t.Parallel() + + diff := BlocksDifference{ + RemovedOrganizations: []string{"org1"}, + } + + // disabled drops the job without touching the inserter. + ir := &insertRecorder{} + require.NoError(t, NewJobs(false).Enqueue(context.Background(), ir, diff)) + assert.Empty(t, ir.diffs) + + // enabled writes through and propagates the inserter's error. + ir = &insertRecorder{err: assert.AnError} + err := NewJobs(true).Enqueue(context.Background(), ir, diff) + assert.Equal(t, assert.AnError, err) + require.Len(t, ir.diffs, 1) + assert.Equal(t, diff, ir.diffs[0]) +} diff --git a/server/core/internal/server/0mock_db_test.go b/server/core/internal/server/0mock_db_test.go index 7e74718c..2f4e40a8 100644 --- a/server/core/internal/server/0mock_db_test.go +++ b/server/core/internal/server/0mock_db_test.go @@ -51,7 +51,7 @@ var _ DB = &DBMock{} // DeleteDataSourceFunc: func(ctx context.Context, id xid.ID, organizationID string) error { // panic("mock out the DeleteDataSource method") // }, -// DeleteDocumentFunc: func(ctx context.Context, id xid.ID, organizationID string) error { +// DeleteDocumentFunc: func(ctx context.Context, id xid.ID, organizationID string) ([]xid.ID, error) { // panic("mock out the DeleteDocument method") // }, // DeleteDocumentBranchByIDFunc: func(ctx context.Context, branchID xid.ID, organizationID string) error { @@ -308,7 +308,7 @@ type DBMock struct { DeleteDataSourceFunc func(ctx context.Context, id xid.ID, organizationID string) error // DeleteDocumentFunc mocks the DeleteDocument method. - DeleteDocumentFunc func(ctx context.Context, id xid.ID, organizationID string) error + DeleteDocumentFunc func(ctx context.Context, id xid.ID, organizationID string) ([]xid.ID, error) // DeleteDocumentBranchByIDFunc mocks the DeleteDocumentBranchByID method. DeleteDocumentBranchByIDFunc func(ctx context.Context, branchID xid.ID, organizationID string) error @@ -1631,7 +1631,7 @@ func (mock *DBMock) DeleteDataSourceCalls() []struct { } // DeleteDocument calls DeleteDocumentFunc. -func (mock *DBMock) DeleteDocument(ctx context.Context, id xid.ID, organizationID string) error { +func (mock *DBMock) DeleteDocument(ctx context.Context, id xid.ID, organizationID string) ([]xid.ID, error) { callInfo := struct { Ctx context.Context ID xid.ID @@ -1646,9 +1646,10 @@ func (mock *DBMock) DeleteDocument(ctx context.Context, id xid.ID, organizationI mock.lockDeleteDocument.Unlock() if mock.DeleteDocumentFunc == nil { var ( + iDsOut []xid.ID errOut error ) - return errOut + return iDsOut, errOut } return mock.DeleteDocumentFunc(ctx, id, organizationID) } diff --git a/server/core/internal/server/capabilities.go b/server/core/internal/server/capabilities.go new file mode 100644 index 00000000..ee2dd1fa --- /dev/null +++ b/server/core/internal/server/capabilities.go @@ -0,0 +1,35 @@ +package server + +import ( + "net/http" + + "github.com/oxynote/oxynote/server/core/pkg/httpserver" +) + +// Capabilities reports which optional services this deployment has +// configured. The frontend reads it to decide what to render at all — +// connect links, the chat entry point, hook types, search — before any +// per-organization state comes into play. +type Capabilities struct { + // GitHub reports whether the GitHub App integration is configured. + GitHub bool `json:"github"` + + // Slack reports whether the Slack App integration is configured. + Slack bool `json:"slack"` + + // Assistant reports whether the AI assistant is configured. + Assistant bool `json:"assistant"` + + // Changedetection reports whether the changedetection.io integration + // is configured. + Changedetection bool `json:"changedetection"` + + // Search reports whether document search is configured. + Search bool `json:"search"` +} + +// fetchCapabilities responds with the deployment's capabilities. The +// values are fixed at boot, so they are snapshotted once in NewServer. +func (s *Server) fetchCapabilities(w http.ResponseWriter, _ *http.Request) { + httpserver.Respond(s.log, w, s.capabilities, http.StatusOK) +} diff --git a/server/core/internal/server/capabilities_test.go b/server/core/internal/server/capabilities_test.go new file mode 100644 index 00000000..cb4e51a7 --- /dev/null +++ b/server/core/internal/server/capabilities_test.go @@ -0,0 +1,33 @@ +package server + +import ( + "net/http" + "net/http/httptest" + "testing" + + "github.com/stretchr/testify/assert" +) + +func Test_Server_fetchCapabilities(t *testing.T) { + t.Parallel() + + s := Server{ + log: discardLog(), + capabilities: Capabilities{ + Slack: true, + Search: true, + }, + } + + req := httptest.NewRequest(http.MethodGet, "http://test.com/", http.NoBody) + rec := httptest.NewRecorder() + + s.fetchCapabilities(rec, req) + + assert.Equal(t, http.StatusOK, rec.Code) + assert.JSONEq( + t, + `{"github":false,"slack":true,"assistant":false,"changedetection":false,"search":true}`, + rec.Body.String(), + ) +} diff --git a/server/core/internal/server/internal/assistant/0mock_manager_test.go b/server/core/internal/server/internal/assistant/0mock_manager_test.go index e23cbb73..003b4768 100644 --- a/server/core/internal/server/internal/assistant/0mock_manager_test.go +++ b/server/core/internal/server/internal/assistant/0mock_manager_test.go @@ -30,11 +30,17 @@ var _ Manager = &ManagerMock{} // // } type ManagerMock struct { + // ConfiguredFunc mocks the Configured method. + ConfiguredFunc func() bool + // ChatFunc mocks the Chat method. ChatFunc func(ctx context.Context, orgID string, userID string, conn protocol.SessionConn) error // calls tracks calls to the methods. calls struct { + // Configured holds details about calls to the Configured method. + Configured []struct { + } // Chat holds details about calls to the Chat method. Chat []struct { // Ctx is the ctx argument value. @@ -47,7 +53,8 @@ type ManagerMock struct { Conn protocol.SessionConn } } - lockChat sync.RWMutex + lockConfigured sync.RWMutex + lockChat sync.RWMutex } // Chat calls ChatFunc. @@ -96,3 +103,33 @@ func (mock *ManagerMock) ChatCalls() []struct { mock.lockChat.RUnlock() return calls } + +// Configured calls ConfiguredFunc. +func (mock *ManagerMock) Configured() bool { + callInfo := struct { + }{} + mock.lockConfigured.Lock() + mock.calls.Configured = append(mock.calls.Configured, callInfo) + mock.lockConfigured.Unlock() + if mock.ConfiguredFunc == nil { + var ( + bOut bool + ) + return bOut + } + return mock.ConfiguredFunc() +} + +// ConfiguredCalls gets all the calls that were made to Configured. +// Check the length with: +// +// len(mockedManager.ConfiguredCalls()) +func (mock *ManagerMock) ConfiguredCalls() []struct { +} { + var calls []struct { + } + mock.lockConfigured.RLock() + calls = mock.calls.Configured + mock.lockConfigured.RUnlock() + return calls +} diff --git a/server/core/internal/server/internal/assistant/http.go b/server/core/internal/server/internal/assistant/http.go index a82909e0..9a5f500c 100644 --- a/server/core/internal/server/internal/assistant/http.go +++ b/server/core/internal/server/internal/assistant/http.go @@ -12,6 +12,7 @@ import ( "github.com/coder/websocket" "github.com/coder/websocket/wsjson" + assistantCore "github.com/oxynote/oxynote/server/core/internal/assistant" "github.com/oxynote/oxynote/server/core/internal/assistant/protocol" "github.com/oxynote/oxynote/server/core/internal/server/internal/auth" "github.com/oxynote/oxynote/server/core/pkg/httpserver" @@ -43,6 +44,13 @@ func NewHandler( // HandleChat upgrades the connection to WebSocket and hands it to the // assistant, which owns the chat loop until the client disconnects. func (h *Handler) HandleChat(w http.ResponseWriter, r *http.Request) { + // refused before the upgrade so the client gets a readable status + // and code instead of a failed websocket handshake. + if !h.assistant.Configured() { + httpserver.RespondError(h.log, w, assistantCore.ErrNotConfigured) + return + } + sess, err := auth.ExtractSessionFromContext(r.Context()) if err != nil { httpserver.RespondError(h.log, w, err) @@ -130,6 +138,10 @@ func (w *wsConn) WriteJSON(ctx context.Context, msg any) { // //go:generate ../../../../scripts/codegen/mock -t internal Manager type Manager interface { + // Configured should report whether the assistant is configured on + // this deployment. + Configured() bool + // Chat should run the chat loop over the connection until the // client disconnects, returning nil on a clean close. Chat(ctx context.Context, orgID, userID string, conn protocol.SessionConn) error diff --git a/server/core/internal/server/internal/assistant/http_test.go b/server/core/internal/server/internal/assistant/http_test.go index 0059eead..61bddf4c 100644 --- a/server/core/internal/server/internal/assistant/http_test.go +++ b/server/core/internal/server/internal/assistant/http_test.go @@ -90,6 +90,15 @@ func Test_Handler_HandleChat(t *testing.T) { } cc := map[string]tcase{ + "Assistant not configured": { + Man: &ManagerMock{ + ConfiguredFunc: func() bool { return false }, + }, + Resp: func(t *testing.T, rec *httptest.ResponseRecorder) { + assert.Equal(t, http.StatusConflict, rec.Code) + assert.JSONEq(t, `{"code":"assistant.not_configured","message":"assistant is not configured"}`, rec.Body.String()) + }, + }, "Unauthenticated request": { Man: &ManagerMock{}, NoSession: true, @@ -210,6 +219,12 @@ func Test_Handler_HandleChat(t *testing.T) { t.Run(cn, func(t *testing.T) { t.Parallel() + // the gate runs before everything else, so every case that + // exercises what follows needs a configured assistant. + if c.Man.ConfiguredFunc == nil { + c.Man.ConfiguredFunc = func() bool { return true } + } + hdl := &Handler{ log: slog.New(slog.DiscardHandler), assistant: c.Man, diff --git a/server/core/internal/server/internal/document/0mock_db_test.go b/server/core/internal/server/internal/document/0mock_db_test.go index f06d808c..3cffc840 100644 --- a/server/core/internal/server/internal/document/0mock_db_test.go +++ b/server/core/internal/server/internal/document/0mock_db_test.go @@ -43,7 +43,7 @@ var _ DB = &DBMock{} // DeleteBranchReviewerFunc: func(ctx context.Context, branchID xid.ID, userID string, organizationID string) error { // panic("mock out the DeleteBranchReviewer method") // }, -// DeleteDocumentFunc: func(ctx context.Context, id xid.ID, organizationID string) error { +// DeleteDocumentFunc: func(ctx context.Context, id xid.ID, organizationID string) ([]xid.ID, error) { // panic("mock out the DeleteDocument method") // }, // DeleteDocumentBranchByIDFunc: func(ctx context.Context, branchID xid.ID, organizationID string) error { @@ -159,7 +159,7 @@ type DBMock struct { DeleteBranchReviewerFunc func(ctx context.Context, branchID xid.ID, userID string, organizationID string) error // DeleteDocumentFunc mocks the DeleteDocument method. - DeleteDocumentFunc func(ctx context.Context, id xid.ID, organizationID string) error + DeleteDocumentFunc func(ctx context.Context, id xid.ID, organizationID string) ([]xid.ID, error) // DeleteDocumentBranchByIDFunc mocks the DeleteDocumentBranchByID method. DeleteDocumentBranchByIDFunc func(ctx context.Context, branchID xid.ID, organizationID string) error @@ -875,7 +875,7 @@ func (mock *DBMock) DeleteBranchReviewerCalls() []struct { } // DeleteDocument calls DeleteDocumentFunc. -func (mock *DBMock) DeleteDocument(ctx context.Context, id xid.ID, organizationID string) error { +func (mock *DBMock) DeleteDocument(ctx context.Context, id xid.ID, organizationID string) ([]xid.ID, error) { callInfo := struct { Ctx context.Context ID xid.ID @@ -890,9 +890,10 @@ func (mock *DBMock) DeleteDocument(ctx context.Context, id xid.ID, organizationI mock.lockDeleteDocument.Unlock() if mock.DeleteDocumentFunc == nil { var ( + iDsOut []xid.ID errOut error ) - return errOut + return iDsOut, errOut } return mock.DeleteDocumentFunc(ctx, id, organizationID) } diff --git a/server/core/internal/server/internal/document/0mock_search_gateway_test.go b/server/core/internal/server/internal/document/0mock_search_gateway_test.go index 3eb4c0d0..21c0c6e6 100644 --- a/server/core/internal/server/internal/document/0mock_search_gateway_test.go +++ b/server/core/internal/server/internal/document/0mock_search_gateway_test.go @@ -28,11 +28,17 @@ var _ SearchGateway = &SearchGatewayMock{} // // } type SearchGatewayMock struct { + // ConfiguredFunc mocks the Configured method. + ConfiguredFunc func() bool + // SearchDocumentsFunc mocks the SearchDocuments method. SearchDocumentsFunc func(ctx context.Context, organizationID string, query string) ([]byte, error) // calls tracks calls to the methods. calls struct { + // Configured holds details about calls to the Configured method. + Configured []struct { + } // SearchDocuments holds details about calls to the SearchDocuments method. SearchDocuments []struct { // Ctx is the ctx argument value. @@ -43,6 +49,7 @@ type SearchGatewayMock struct { Query string } } + lockConfigured sync.RWMutex lockSearchDocuments sync.RWMutex } @@ -89,3 +96,33 @@ func (mock *SearchGatewayMock) SearchDocumentsCalls() []struct { mock.lockSearchDocuments.RUnlock() return calls } + +// Configured calls ConfiguredFunc. +func (mock *SearchGatewayMock) Configured() bool { + callInfo := struct { + }{} + mock.lockConfigured.Lock() + mock.calls.Configured = append(mock.calls.Configured, callInfo) + mock.lockConfigured.Unlock() + if mock.ConfiguredFunc == nil { + var ( + bOut bool + ) + return bOut + } + return mock.ConfiguredFunc() +} + +// ConfiguredCalls gets all the calls that were made to Configured. +// Check the length with: +// +// len(mockedSearchGateway.ConfiguredCalls()) +func (mock *SearchGatewayMock) ConfiguredCalls() []struct { +} { + var calls []struct { + } + mock.lockConfigured.RLock() + calls = mock.calls.Configured + mock.lockConfigured.RUnlock() + return calls +} diff --git a/server/core/internal/server/internal/document/0mock_tx_test.go b/server/core/internal/server/internal/document/0mock_tx_test.go index e18b81ab..f15a10a3 100644 --- a/server/core/internal/server/internal/document/0mock_tx_test.go +++ b/server/core/internal/server/internal/document/0mock_tx_test.go @@ -43,7 +43,7 @@ var _ Tx = &TxMock{} // DeleteBranchReviewerFunc: func(ctx context.Context, branchID xid.ID, userID string, organizationID string) error { // panic("mock out the DeleteBranchReviewer method") // }, -// DeleteDocumentFunc: func(ctx context.Context, id xid.ID, organizationID string) error { +// DeleteDocumentFunc: func(ctx context.Context, id xid.ID, organizationID string) ([]xid.ID, error) { // panic("mock out the DeleteDocument method") // }, // DeleteDocumentBranchByIDFunc: func(ctx context.Context, branchID xid.ID, organizationID string) error { @@ -162,7 +162,7 @@ type TxMock struct { DeleteBranchReviewerFunc func(ctx context.Context, branchID xid.ID, userID string, organizationID string) error // DeleteDocumentFunc mocks the DeleteDocument method. - DeleteDocumentFunc func(ctx context.Context, id xid.ID, organizationID string) error + DeleteDocumentFunc func(ctx context.Context, id xid.ID, organizationID string) ([]xid.ID, error) // DeleteDocumentBranchByIDFunc mocks the DeleteDocumentBranchByID method. DeleteDocumentBranchByIDFunc func(ctx context.Context, branchID xid.ID, organizationID string) error @@ -872,7 +872,7 @@ func (mock *TxMock) DeleteBranchReviewerCalls() []struct { } // DeleteDocument calls DeleteDocumentFunc. -func (mock *TxMock) DeleteDocument(ctx context.Context, id xid.ID, organizationID string) error { +func (mock *TxMock) DeleteDocument(ctx context.Context, id xid.ID, organizationID string) ([]xid.ID, error) { callInfo := struct { Ctx context.Context ID xid.ID @@ -887,9 +887,10 @@ func (mock *TxMock) DeleteDocument(ctx context.Context, id xid.ID, organizationI mock.lockDeleteDocument.Unlock() if mock.DeleteDocumentFunc == nil { var ( + iDsOut []xid.ID errOut error ) - return errOut + return iDsOut, errOut } return mock.DeleteDocumentFunc(ctx, id, organizationID) } diff --git a/server/core/internal/server/internal/document/branch_http_test.go b/server/core/internal/server/internal/document/branch_http_test.go index 6968793c..ecfc5945 100644 --- a/server/core/internal/server/internal/document/branch_http_test.go +++ b/server/core/internal/server/internal/document/branch_http_test.go @@ -602,7 +602,19 @@ func Test_Handler_DeleteDocument(t *testing.T) { "Document deletion error": { DB: &DBMock{FetchDocumentFunc: fetchStored}, Tx: &TxMock{ - DeleteDocumentFunc: func(context.Context, xid.ID, string) error { + DeleteDocumentFunc: func(context.Context, xid.ID, string) ([]xid.ID, error) { + return nil, errors.New("boom") + }, + }, + RespCode: http.StatusInternalServerError, + }, + "Search removal enqueue error": { + DB: &DBMock{FetchDocumentFunc: fetchStored}, + Tx: &TxMock{ + DeleteDocumentFunc: func(context.Context, xid.ID, string) ([]xid.ID, error) { + return []xid.ID{_documentID}, nil + }, + InsertDocumentSearchJobFunc: func(context.Context, search.BlocksDifference) error { return errors.New("boom") }, }, @@ -625,7 +637,11 @@ func Test_Handler_DeleteDocument(t *testing.T) { return []hookCore.Hook{storedHook(hookCore.TypeScheduledReminder)}, nil }, }, - Tx: &TxMock{}, + Tx: &TxMock{ + DeleteDocumentFunc: func(_ context.Context, id xid.ID, _ string) ([]xid.ID, error) { + return []xid.ID{id}, nil + }, + }, RespCode: http.StatusOK, Committed: 1, TreeCbs: 1, @@ -650,9 +666,11 @@ func Test_Handler_DeleteDocument(t *testing.T) { require.Len(t, c.Tx.DeleteDocumentCalls(), 1) assert.Equal(t, _documentID, c.Tx.DeleteDocumentCalls()[0].ID) - // the search removal is queued by DeleteDocument itself, - // which is what makes it cover the cascaded descendants. - assert.Empty(t, c.Tx.InsertDocumentSearchJobCalls()) + // the removal of the returned subtree ids rides the same + // transaction as the delete. + ff := c.Tx.InsertDocumentSearchJobCalls() + require.Len(t, ff, 1) + assert.Equal(t, []xid.ID{_documentID}, ff[0].Diff.RemovedDocuments) } }) } @@ -1278,6 +1296,33 @@ func Test_Handler_DeleteDocumentBranch(t *testing.T) { func Test_Handler_copyHooksToBranch(t *testing.T) { t.Parallel() + // a url-watcher cannot get its watcher without changedetection + // configured, so the copy drops it and still copies the rest. + t.Run("URL watcher skipped without changedetection", func(t *testing.T) { + t.Parallel() + + urlHook := storedHook(hookCore.TypeURLWatcher) + urlHook.Settings = processor.Settings(`{"url":"https://example.com"}`) + + db := &DBMock{ + FetchDocumentHooksByBranchIDFunc: func(context.Context, xid.ID, string) ([]hookCore.Hook, error) { + return []hookCore.Hook{urlHook, storedHook(hookCore.TypeScheduledReminder)}, nil + }, + InsertDocumentHookFunc: func(context.Context, hookCore.Hook) error { + return nil + }, + } + + hdl, _ := newTestHandler(db, &fakePublisher{}) + + err := hdl.copyHooksToBranch(context.Background(), _branchID2, _branchID, _documentID, "org1") + require.NoError(t, err) + + ff := db.InsertDocumentHookCalls() + require.Len(t, ff, 1) + assert.Equal(t, hookCore.TypeScheduledReminder, ff[0].Hk.Type) + }) + // creating a hook creates its external resource, so a failed insert has // to hand it back rather than leave it running with no row pointing at // it. A url-watcher would call changedetection.io here, which the test diff --git a/server/core/internal/server/internal/document/hook/http_test.go b/server/core/internal/server/internal/document/hook/http_test.go index fb011e45..c976be78 100644 --- a/server/core/internal/server/internal/document/hook/http_test.go +++ b/server/core/internal/server/internal/document/hook/http_test.go @@ -10,6 +10,7 @@ import ( "testing" "github.com/guregu/null/v5" + "github.com/oxynote/oxynote/server/core/internal/apps/webchange" "github.com/oxynote/oxynote/server/core/internal/document" hookCore "github.com/oxynote/oxynote/server/core/internal/document/hook" "github.com/oxynote/oxynote/server/core/internal/document/hook/processor" @@ -40,6 +41,19 @@ func addSession(ctx context.Context) context.Context { }) } +// urlWatcherHook builds a stored url-watcher hook that already holds a +// changedetection.io watcher in its state. +func urlWatcherHook() *hookCore.Hook { + return &hookCore.Hook{ + ID: _hookID, + Type: hookCore.TypeURLWatcher, + DocumentID: null.ValueFrom(_documentID), + OrganizationID: null.StringFrom("org1"), + Settings: processor.Settings(`{"url":"https://example.com"}`), + State: processor.State(`{"watcherId":"w1"}`), + } +} + // scheduledHook builds a stored hook whose processor needs no external // dependencies. func scheduledHook(typ hookCore.Type) *hookCore.Hook { @@ -154,8 +168,9 @@ func Test_Handler_FetchDocumentHooks(t *testing.T) { t.Parallel() hdl := Handler{ - log: slog.New(slog.DiscardHandler), - db: c.DB, + log: slog.New(slog.DiscardHandler), + db: c.DB, + webchangeClient: webchange.NewClient("", ""), } req := httptest.NewRequest(http.MethodGet, "http://test.com/"+c.Query, http.NoBody) @@ -288,6 +303,14 @@ func Test_Handler_CreateDocumentHook(t *testing.T) { wasInsertCalled(0), ), }, + "URL watcher without changedetection": { + DB: &DBMock{}, + Body: `{"type":"url-watcher","branchId":"` + _branchID.String() + `","settings":{"url":"https://example.com"}}`, + Checks: checks( + hasResp(http.StatusConflict, `{"code":"changedetection.not_configured","message":"changedetection is not configured"}`), + wasInsertCalled(0), + ), + }, "Successful creation": { DB: &DBMock{}, Body: validBody, @@ -312,8 +335,9 @@ func Test_Handler_CreateDocumentHook(t *testing.T) { } hdl := Handler{ - log: slog.New(slog.DiscardHandler), - db: c.DB, + log: slog.New(slog.DiscardHandler), + db: c.DB, + webchangeClient: webchange.NewClient("", ""), } req := httptest.NewRequest(http.MethodPost, "http://test.com/", strings.NewReader(c.Body)) @@ -450,6 +474,18 @@ func Test_Handler_UpdateDocumentHook(t *testing.T) { wasUpdateCalled(1), ), }, + "URL watcher without changedetection": { + DB: &DBMock{ + FetchDocumentHookFunc: func(context.Context, xid.ID, string) (*hookCore.Hook, error) { + return urlWatcherHook(), nil + }, + }, + Body: `{"settings":{"url":"https://example.com"}}`, + Checks: checks( + hasResp(http.StatusConflict, `{"code":"changedetection.not_configured","message":"changedetection is not configured"}`), + wasUpdateCalled(0), + ), + }, "Successful update": { DB: &DBMock{ FetchDocumentHookFunc: func(context.Context, xid.ID, string) (*hookCore.Hook, error) { @@ -472,8 +508,9 @@ func Test_Handler_UpdateDocumentHook(t *testing.T) { t.Parallel() hdl := Handler{ - log: slog.New(slog.DiscardHandler), - db: c.DB, + log: slog.New(slog.DiscardHandler), + db: c.DB, + webchangeClient: webchange.NewClient("", ""), } req := httptest.NewRequest(http.MethodPut, "http://test.com/", strings.NewReader(c.Body)) @@ -567,6 +604,15 @@ func Test_Handler_ResetDocumentHook(t *testing.T) { RespCode: http.StatusInternalServerError, Checks: checks(wasUpdateCalled(1)), }, + "URL watcher without changedetection": { + DB: &DBMock{ + FetchDocumentHookFunc: func(context.Context, xid.ID, string) (*hookCore.Hook, error) { + return urlWatcherHook(), nil + }, + }, + RespCode: http.StatusConflict, + Checks: checks(wasUpdateCalled(0)), + }, "Successful reset": { DB: &DBMock{ FetchDocumentHookFunc: func(context.Context, xid.ID, string) (*hookCore.Hook, error) { @@ -583,8 +629,9 @@ func Test_Handler_ResetDocumentHook(t *testing.T) { t.Parallel() hdl := Handler{ - log: slog.New(slog.DiscardHandler), - db: c.DB, + log: slog.New(slog.DiscardHandler), + db: c.DB, + webchangeClient: webchange.NewClient("", ""), } req := httptest.NewRequest(http.MethodPost, "http://test.com/", http.NoBody) @@ -688,6 +735,18 @@ func Test_Handler_DeleteDocumentHook(t *testing.T) { RespCode: http.StatusNoContent, Checks: checks(wasDeleteCalled(1)), }, + + // an unconfigured changedetection must not trap the row: the + // teardown is skipped and the deletion proceeds. + "URL watcher deletion without changedetection": { + DB: &DBMock{ + FetchDocumentHookFunc: func(context.Context, xid.ID, string) (*hookCore.Hook, error) { + return urlWatcherHook(), nil + }, + }, + RespCode: http.StatusNoContent, + Checks: checks(wasDeleteCalled(1)), + }, } for cn, c := range cc { @@ -695,8 +754,9 @@ func Test_Handler_DeleteDocumentHook(t *testing.T) { t.Parallel() hdl := Handler{ - log: slog.New(slog.DiscardHandler), - db: c.DB, + log: slog.New(slog.DiscardHandler), + db: c.DB, + webchangeClient: webchange.NewClient("", ""), } req := httptest.NewRequest(http.MethodDelete, "http://test.com/", http.NoBody) diff --git a/server/core/internal/server/internal/document/http.go b/server/core/internal/server/internal/document/http.go index 7bf8a4a9..d6286345 100644 --- a/server/core/internal/server/internal/document/http.go +++ b/server/core/internal/server/internal/document/http.go @@ -45,6 +45,7 @@ type Handler struct { githubMan *github.Manager webchangeClient *webchange.Client searchGateway SearchGateway + searchJobs *search.Jobs notifPub notification.Publisher storer Storer @@ -72,6 +73,7 @@ func NewHandler( githubMan *github.Manager, webchangeClient *webchange.Client, searchGateway SearchGateway, + searchJobs *search.Jobs, notifPub notification.Publisher, storer Storer, ) *Handler { @@ -81,6 +83,7 @@ func NewHandler( githubMan: githubMan, webchangeClient: webchangeClient, searchGateway: searchGateway, + searchJobs: searchJobs, notifPub: notifPub, storer: storer, } @@ -664,8 +667,9 @@ func (h *Handler) UpdateDocumentBranchByIDUnsafe(w http.ResponseWriter, r *http. } if ndoc.BranchName == documentCore.DefaultBranch { - if err = tx.InsertDocumentSearchJob( + if err = h.searchJobs.Enqueue( r.Context(), + tx, search.BlocksDiff(doc.Search(), ndoc.Search()), ); err != nil { httpserver.RespondError(h.log, w, err) @@ -817,7 +821,7 @@ func (h *Handler) MergeBranches(w http.ResponseWriter, r *http.Request) { return } - if err := tx.InsertDocumentSearchJob(r.Context(), search.BlocksDiff(toDoc.Search(), ndoc.Search())); err != nil { + if err := h.searchJobs.Enqueue(r.Context(), tx, search.BlocksDiff(toDoc.Search(), ndoc.Search())); err != nil { httpserver.RespondError(h.log, w, err) return } @@ -911,13 +915,25 @@ func (h *Handler) DeleteDocument(w http.ResponseWriter, r *http.Request) { defer tx.Rollback() //nolint:errcheck // error provides no meaningful info - // the search index entries of the document and its cascade-deleted - // descendants are queued for removal by DeleteDocument itself. - if err = tx.DeleteDocument(r.Context(), doc.ID, session.ActiveOrganizationID); err != nil { + // the delete reports the ids of the document and its cascade-deleted + // descendants; queuing their index removal from them in the same + // transaction is this handler's job, since after the commit nothing + // else knows what went away. + ids, err := tx.DeleteDocument(r.Context(), doc.ID, session.ActiveOrganizationID) + if err != nil { httpserver.RespondError(h.log, w, err) return } + if len(ids) != 0 { + if err = h.searchJobs.Enqueue(r.Context(), tx, search.BlocksDifference{ + RemovedDocuments: ids, + }); err != nil { + httpserver.RespondError(h.log, w, err) + return + } + } + err = tx.Commit() if err != nil { httpserver.RespondError(h.log, w, err) @@ -1216,6 +1232,16 @@ func (h *Handler) copyHooksToBranch(ctx context.Context, fromBranchID, toBranchI inp := hook.NewInput(organizationID, h.githubMan, h.webchangeClient) for _, hk := range hooks { + // a url-watcher cannot get its changedetection.io watcher without + // the integration configured; dropping it from the copy beats + // failing the whole fork or merge over it. + if hk.Type == hook.TypeURLWatcher && !h.webchangeClient.Configured() { + h.log.With("hook_id", hk.ID). + Warn("skipping url-watcher hook copy: changedetection is not configured") + + continue + } + newHk, err := hook.NewHook(ctx, hook.CreateInput{ Type: hk.Type, BranchID: toBranchID, @@ -1303,7 +1329,7 @@ func (h *Handler) insertDocumentTx(ctx context.Context, doc documentCore.Documen return err } - if err := tx.InsertDocumentSearchJob(ctx, search.BlocksDiff(nil, doc.Search())); err != nil { + if err := h.searchJobs.Enqueue(ctx, tx, search.BlocksDiff(nil, doc.Search())); err != nil { return err } @@ -1404,8 +1430,9 @@ type DocumentsDBAgent interface { // UpdateDocument should update the document. UpdateDocument(ctx context.Context, doc documentCore.Document) error - // DeleteDocument should delete the document. - DeleteDocument(ctx context.Context, id xid.ID, organizationID string) error + // DeleteDocument should delete the document and report the ids of + // the document and of every cascade-deleted descendant. + DeleteDocument(ctx context.Context, id xid.ID, organizationID string) ([]xid.ID, error) } // BranchesDBAgent is an interface that handles communication with the @@ -1507,6 +1534,10 @@ type Storer interface { // //go:generate ../../../../scripts/codegen/mock -t internal SearchGateway type SearchGateway interface { + // Configured should report whether search is configured on this + // deployment. + Configured() bool + // SearchDocuments should find the documents matching the query. SearchDocuments(ctx context.Context, organizationID, query string) ([]byte, error) } diff --git a/server/core/internal/server/internal/document/http_test.go b/server/core/internal/server/internal/document/http_test.go index 67d32520..7045be7f 100644 --- a/server/core/internal/server/internal/document/http_test.go +++ b/server/core/internal/server/internal/document/http_test.go @@ -11,6 +11,7 @@ import ( "testing" "github.com/guregu/null/v5" + "github.com/oxynote/oxynote/server/core/internal/apps/webchange" documentCore "github.com/oxynote/oxynote/server/core/internal/document" "github.com/oxynote/oxynote/server/core/internal/notification" "github.com/oxynote/oxynote/server/core/internal/search" @@ -62,10 +63,12 @@ func newTestHandler(db DB, pub *fakePublisher) (*Handler, *callbackCounts) { cnt := &callbackCounts{} hdl := &Handler{ - log: slog.New(slog.DiscardHandler), - db: db, - notifPub: pub, - storer: &StorerMock{}, + log: slog.New(slog.DiscardHandler), + db: db, + notifPub: pub, + storer: &StorerMock{}, + webchangeClient: webchange.NewClient("", ""), + searchJobs: search.NewJobs(true), } hdl.tree.changeCallback = func(string, null.Value[xid.ID]) { cnt.tree++ } @@ -148,14 +151,16 @@ func Test_NewHandler(t *testing.T) { db := &DBMock{} gw := &SearchGatewayMock{} + jobs := search.NewJobs(true) pub := &fakePublisher{} st := &StorerMock{} - hdl := NewHandler(slog.New(slog.DiscardHandler), db, nil, nil, gw, pub, st) + hdl := NewHandler(slog.New(slog.DiscardHandler), db, nil, nil, gw, jobs, pub, st) require.NotNil(t, hdl) assert.NotNil(t, hdl.log) assert.Same(t, db, hdl.db) assert.Same(t, gw, hdl.searchGateway) + assert.Same(t, jobs, hdl.searchJobs) assert.Same(t, pub, hdl.notifPub) assert.Same(t, st, hdl.storer) assert.Nil(t, hdl.githubMan) @@ -972,6 +977,16 @@ func Test_Handler_SearchDocuments(t *testing.T) { Query: "?q=test", RespCode: http.StatusInternalServerError, }, + "Search not configured": { + Gateway: &SearchGatewayMock{ + SearchDocumentsFunc: func(context.Context, string, string) ([]byte, error) { + return nil, search.ErrNotConfigured + }, + }, + Query: "?q=test", + RespCode: http.StatusConflict, + RespBody: `{"code":"search.not_configured","message":"search is not configured"}`, + }, "Successful search": { Gateway: &SearchGatewayMock{ SearchDocumentsFunc: func(context.Context, string, string) ([]byte, error) { diff --git a/server/core/internal/server/internal/mcp/http_test.go b/server/core/internal/server/internal/mcp/http_test.go index 996e85ce..0cd9709f 100644 --- a/server/core/internal/server/internal/mcp/http_test.go +++ b/server/core/internal/server/internal/mcp/http_test.go @@ -12,6 +12,7 @@ import ( "github.com/jarcoal/httpmock" "github.com/oxynote/oxynote/server/core/internal/assistant/tools" toolsMock "github.com/oxynote/oxynote/server/core/internal/assistant/tools/_mock" + "github.com/oxynote/oxynote/server/core/internal/search" "github.com/oxynote/oxynote/server/core/pkg/testutil" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -46,7 +47,10 @@ func stubSet(db *toolsMock.DB) *tools.Set { return tools.New(tools.NewDeps( discardLog(), db, - &toolsMock.Searcher{}, + &toolsMock.Searcher{ + ConfiguredFunc: func() bool { return true }, + }, + search.NewJobs(true), nil, &toolsMock.EditApplier{}, nil, diff --git a/server/core/internal/server/internal/org/http.go b/server/core/internal/server/internal/org/http.go index 8f0c5bc0..475758d4 100644 --- a/server/core/internal/server/internal/org/http.go +++ b/server/core/internal/server/internal/org/http.go @@ -37,6 +37,7 @@ type Handler struct { storer Storer githubMan *github.Manager webchangeClient *webchange.Client + searchJobs *search.Jobs logoLocation string demoPrometheusURL string } @@ -48,6 +49,7 @@ func NewHandler( storer Storer, githubMan *github.Manager, webchangeClient *webchange.Client, + searchJobs *search.Jobs, logoLocationFormat string, demoPrometheusURL string, ) *Handler { @@ -57,6 +59,7 @@ func NewHandler( storer: storer, githubMan: githubMan, webchangeClient: webchangeClient, + searchJobs: searchJobs, logoLocation: logoLocationFormat, demoPrometheusURL: demoPrometheusURL, } @@ -134,7 +137,7 @@ func (h *Handler) InitializeOrganization(w http.ResponseWriter, r *http.Request) return } - if err = tx.InsertDocumentSearchJob(r.Context(), search.BlocksDiff(nil, doc.Search())); err != nil { + if err = h.searchJobs.Enqueue(r.Context(), tx, search.BlocksDiff(nil, doc.Search())); err != nil { httpserver.RespondError(h.log, w, err) return } @@ -270,7 +273,7 @@ func (h *Handler) TeardownOrganization(w http.ResponseWriter, r *http.Request) { defer tx.Rollback() //nolint:errcheck // error provides no meaningful info - if err = tx.InsertDocumentSearchJob(r.Context(), search.BlocksDifference{ + if err = h.searchJobs.Enqueue(r.Context(), tx, search.BlocksDifference{ RemovedOrganizations: []string{id}, }); err != nil { httpserver.RespondError(h.log, w, err) diff --git a/server/core/internal/server/internal/org/http_test.go b/server/core/internal/server/internal/org/http_test.go index bd02676f..b3949b1e 100644 --- a/server/core/internal/server/internal/org/http_test.go +++ b/server/core/internal/server/internal/org/http_test.go @@ -83,12 +83,15 @@ func Test_NewHandler(t *testing.T) { githubMan := &github.Manager{} webchangeClient := &webchange.Client{} + searchJobs := search.NewJobs(true) + hdl := NewHandler( slog.New(slog.DiscardHandler), db, storer, githubMan, webchangeClient, + searchJobs, "loc", "http://prom.test", ) @@ -98,6 +101,7 @@ func Test_NewHandler(t *testing.T) { assert.Same(t, storer, hdl.storer) assert.Same(t, githubMan, hdl.githubMan) assert.Same(t, webchangeClient, hdl.webchangeClient) + assert.Same(t, searchJobs, hdl.searchJobs) assert.Equal(t, "loc", hdl.logoLocation) assert.Equal(t, "http://prom.test", hdl.demoPrometheusURL) } @@ -355,6 +359,7 @@ func Test_Handler_InitializeOrganization(t *testing.T) { hdl := Handler{ log: slog.New(slog.DiscardHandler), db: withTx(db, c.Tx, c.BeginErr), + searchJobs: search.NewJobs(true), demoPrometheusURL: promURL, } @@ -915,9 +920,10 @@ func Test_Handler_TeardownOrganization(t *testing.T) { } hdl := Handler{ - log: slog.New(slog.DiscardHandler), - db: withTx(c.DB, c.Tx, c.BeginErr), - storer: storer, + log: slog.New(slog.DiscardHandler), + db: withTx(c.DB, c.Tx, c.BeginErr), + storer: storer, + searchJobs: search.NewJobs(true), } req := httptest.NewRequest(http.MethodPost, "http://test.com/", http.NoBody) diff --git a/server/core/internal/server/router.go b/server/core/internal/server/router.go index 9d9836e7..5174a6f9 100644 --- a/server/core/internal/server/router.go +++ b/server/core/internal/server/router.go @@ -183,6 +183,8 @@ func (s *Server) router() chi.Router { s.ws.ServeHTTP(w, r) })) + r.Get("/capabilities", s.fetchCapabilities) + r.Route("/github", func(sr chi.Router) { // the status endpoint stays available when the GitHub App is not // configured: it is the capability signal for the frontend. diff --git a/server/core/internal/server/router_test.go b/server/core/internal/server/router_test.go index 8cdb89b6..5a4c435b 100644 --- a/server/core/internal/server/router_test.go +++ b/server/core/internal/server/router_test.go @@ -74,6 +74,10 @@ func Test_Server_httpRouter(t *testing.T) { assert.False(t, routes[route], "%s sits behind the front door's 403", route) } + // the capability signal lives on the session-authed surface (the "/" + // mount walks as "*"). + assert.True(t, routes["GET /api/*/capabilities"]) + // the service-to-service surface stays where it is. assert.True(t, routes["POST /api/x/organizations/{organizationId}/teardown"]) assert.True(t, routes["PUT /api/x/documents/{documentId}/branch/{branchId}/"]) diff --git a/server/core/internal/server/server.go b/server/core/internal/server/server.go index d8b84d55..cbbe601a 100644 --- a/server/core/internal/server/server.go +++ b/server/core/internal/server/server.go @@ -21,6 +21,7 @@ import ( datasourceCore "github.com/oxynote/oxynote/server/core/internal/datasource" documentCore "github.com/oxynote/oxynote/server/core/internal/document" notificationCore "github.com/oxynote/oxynote/server/core/internal/notification" + "github.com/oxynote/oxynote/server/core/internal/search" "github.com/oxynote/oxynote/server/core/internal/server/internal/assistant" "github.com/oxynote/oxynote/server/core/internal/server/internal/auth" "github.com/oxynote/oxynote/server/core/internal/server/internal/datasource" @@ -96,6 +97,11 @@ type Server struct { ws wsserver.Pool + // capabilities is the deployment's optional-service availability, + // snapshotted at construction since configuration cannot change at + // runtime. + capabilities Capabilities + handlers struct { user *user.Handler organization *org.Handler @@ -129,6 +135,7 @@ func NewServer( slackMan *slackCore.Manager, webchangeClient *webchange.Client, searchGateway document.SearchGateway, + searchJobs *search.Jobs, notifier Notifier, emailSender email.Sender, client *http.Client, @@ -142,6 +149,13 @@ func NewServer( opts: opts, fc: fc, client: client, + capabilities: Capabilities{ + GitHub: githubMan.Configured(), + Slack: slackMan.Configured(), + Assistant: assistantMan.Configured(), + Changedetection: webchangeClient.Configured(), + Search: searchGateway.Configured(), + }, } srv.handlers.user = user.NewHandler(log, db, storageClient, opts.PublicURL+_userImageLocationFormat) @@ -151,10 +165,11 @@ func NewServer( storageClient, githubMan, webchangeClient, + searchJobs, opts.PublicURL+_organizationLogoLocation, opts.DemoPrometheusURL, ) - srv.handlers.document = document.NewHandler(log, db, githubMan, webchangeClient, searchGateway, notifier, storageClient) + srv.handlers.document = document.NewHandler(log, db, githubMan, webchangeClient, searchGateway, searchJobs, notifier, storageClient) srv.handlers.comment = comment.NewHandler(log, db, notifier) srv.handlers.files = files.NewHandler(log, db, storageClient, opts.PublicURL+documentCore.FilePathFormat) srv.handlers.hook = hook.NewHandler(log, db, githubMan, webchangeClient) diff --git a/server/core/internal/server/server_test.go b/server/core/internal/server/server_test.go index d72cc7b0..ea8e6b07 100644 --- a/server/core/internal/server/server_test.go +++ b/server/core/internal/server/server_test.go @@ -12,9 +12,11 @@ import ( "github.com/gomodule/redigo/redis" "github.com/oxynote/oxynote/server/core/internal/apps/github" "github.com/oxynote/oxynote/server/core/internal/apps/slack" + "github.com/oxynote/oxynote/server/core/internal/apps/webchange" assistantCore "github.com/oxynote/oxynote/server/core/internal/assistant" "github.com/oxynote/oxynote/server/core/internal/buildinfo" datasourceCore "github.com/oxynote/oxynote/server/core/internal/datasource" + "github.com/oxynote/oxynote/server/core/internal/search" "github.com/oxynote/oxynote/server/core/pkg/metricutil" "github.com/oxynote/oxynote/server/core/pkg/testutil" wsMock "github.com/oxynote/wetsocks/wsserver/_mock" @@ -54,13 +56,27 @@ func Test_Options_validate(t *testing.T) { assert.EqualError(t, Options{}.validate(), "missing mcp session url") } +// stubSearchGateway satisfies the document handler's SearchGateway with a +// fixed configured flag. +type stubSearchGateway struct { + configured bool +} + +func (g stubSearchGateway) Configured() bool { + return g.configured +} + +func (stubSearchGateway) SearchDocuments(context.Context, string, string) ([]byte, error) { + return nil, nil +} + func Test_NewServer(t *testing.T) { t.Parallel() log := discardLog() fc := metricutil.NewFactory("test", prometheus.NewRegistry()) - assistantMan := assistantCore.NewManager(log, nil, &redis.Pool{}, nil, nil, fc, nil, nil, nil, "claude") + assistantMan := assistantCore.NewManager(log, nil, &redis.Pool{}, nil, nil, fc, nil, nil, search.NewJobs(false), nil, "claude") githubMan, err := github.NewManager(nil, github.Options{}) require.NoError(t, err) @@ -85,8 +101,9 @@ func Test_NewServer(t *testing.T) { datasourceCore.NewManager(log, nil), githubMan, slackMan, - nil, - nil, + webchange.NewClient("", ""), + stubSearchGateway{}, + search.NewJobs(false), notifier, nil, http.DefaultClient, @@ -113,12 +130,13 @@ func Test_NewServer(t *testing.T) { // a manager of this subtest's own: NewServer calls // SetTreeNotifier, which must not race the parallel // success case's identical call on a shared manager. - assistantCore.NewManager(log, nil, &redis.Pool{}, nil, nil, fc, nil, nil, nil, "claude"), + assistantCore.NewManager(log, nil, &redis.Pool{}, nil, nil, fc, nil, nil, search.NewJobs(false), nil, "claude"), datasourceCore.NewManager(log, nil), githubMan, slackMan, - nil, - nil, + webchange.NewClient("", ""), + stubSearchGateway{}, + search.NewJobs(false), notifier, nil, http.DefaultClient, @@ -150,8 +168,9 @@ func Test_NewServer(t *testing.T) { datasourceCore.NewManager(log, nil), githubMan, slackMan, - nil, - nil, + webchange.NewClient("", ""), + stubSearchGateway{}, + search.NewJobs(false), notifier, nil, http.DefaultClient, @@ -184,6 +203,50 @@ func Test_NewServer(t *testing.T) { assert.NotNil(t, srv.handlers.email) assert.NotNil(t, srv.handlers.ai) assert.NotNil(t, srv.handlers.mcp) + + // nothing above is configured, so every capability reports false; + // the search gateway mock is the one switched on. + assert.Equal(t, Capabilities{}, srv.capabilities) + }) + + t.Run("Capabilities reflect the configured services", func(t *testing.T) { + t.Parallel() + + srv, err := NewServer( + log, + Options{ + Port: 8080, + MCP: MCPOptions{ + SessionURL: "http://auth.test/api/internal/mcp/session", + ResourceURL: "http://test.com/core/api/mcp", + }, + }, + db, + fc, + storer, + assistantCore.NewManager(log, nil, &redis.Pool{}, nil, nil, fc, nil, nil, search.NewJobs(false), nil, "claude"), + datasourceCore.NewManager(log, nil), + githubMan, + slackMan, + webchange.NewClient("http://changedetection.test", ""), + stubSearchGateway{configured: true}, + search.NewJobs(true), + notifier, + nil, + http.DefaultClient, + ) + require.NoError(t, err) + require.NotNil(t, srv) + + t.Cleanup(func() { + srv.ws.Close() + assert.NoError(t, srv.Close()) + }) + + assert.Equal(t, Capabilities{ + Changedetection: true, + Search: true, + }, srv.capabilities) }) }