diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..1852838 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,72 @@ +name: CI + +on: + pull_request: + push: + branches: [main] + +concurrency: + group: ${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + +env: + CARGO_TERM_COLOR: always + # Debug info only slows linking and bloats the cache in CI. + CARGO_PROFILE_DEV_DEBUG: 0 + +jobs: + lint: + name: fmt and clippy + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: dtolnay/rust-toolchain@stable + with: + components: rustfmt, clippy + - uses: Swatinem/rust-cache@v2 + - run: cargo fmt --all -- --check + - run: cargo clippy --all-targets --locked -- -D warnings + + test: + name: tests + runs-on: ubuntu-latest + services: + # Matches docker-compose.yml; later releases need an auth token. + localstack: + image: localstack/localstack:4.14 + ports: + - 4566:4566 + env: + SERVICES: s3 + options: >- + --health-cmd "curl -sf http://localhost:4566/_localstack/health" + --health-interval 2s + --health-timeout 2s + --health-retries 30 + env: + AWS_ENDPOINT_URL: http://localhost:4566 + AWS_ACCESS_KEY_ID: test + AWS_SECRET_ACCESS_KEY: test + AWS_REGION: us-east-1 + steps: + - uses: actions/checkout@v4 + - uses: dtolnay/rust-toolchain@stable + - uses: Swatinem/rust-cache@v2 + # The embedding models are large and change rarely. + - uses: actions/cache@v4 + with: + path: .fastembed_cache + key: fastembed-${{ hashFiles('src/embed.rs') }} + restore-keys: fastembed- + - name: Unit tests + run: cargo test --locked + # s3vectors skips itself without TEST_VECTOR_BUCKET. + - name: Integration and e2e tests + run: cargo test --locked -- --ignored + - name: Upload e2e server logs + if: failure() + uses: actions/upload-artifact@v4 + with: + name: e2e-logs + path: target/e2e-logs/ + if-no-files-found: ignore diff --git a/.github/workflows/docker.yml b/.github/workflows/docker.yml new file mode 100644 index 0000000..ace638c --- /dev/null +++ b/.github/workflows/docker.yml @@ -0,0 +1,32 @@ +name: Docker + +# A release build plus the baked-in model takes many minutes, so PRs only +# build the image when something that affects it changes. +on: + pull_request: + paths: + - Dockerfile + - .dockerignore + - Cargo.toml + - Cargo.lock + - .github/workflows/docker.yml + push: + branches: [main] + +concurrency: + group: ${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + +jobs: + docker: + name: docker build + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: docker/setup-buildx-action@v3 + - uses: docker/build-push-action@v6 + with: + context: . + push: false + cache-from: type=gha + cache-to: type=gha,mode=max diff --git a/README.md b/README.md index ba0ef1b..8559ca9 100644 --- a/README.md +++ b/README.md @@ -9,12 +9,17 @@ can paste straight into a prompt. AI API. - S3 as the only storage backend, with an optional Amazon S3 Vectors index for larger data. -- Idempotent writes. +- Idempotent writes, and updates and deletes by your own `sourceId`. +- Webhooks that turn events from other services into datapoints, with no + code in between. - Everything scoped to a `projectId`. ## Status -Early but working: everything below is implemented and tested. See +Early but working: everything below is implemented and tested. That is +ingest (single and batch), recall, delete, listing stored datapoints, and +inbound webhooks. There is no auth yet, so run it on a private network. +[Not yet](#not-yet) lists what is missing. See [ARCHITECTURE.md](ARCHITECTURE.md) for how it works and [CONTRIBUTING.md](CONTRIBUTING.md) to run it locally. @@ -34,7 +39,16 @@ cp .env.example .env # point the AWS SDK at LocalStack cargo run # first run downloads the embedding model ``` -The service listens on `http://localhost:8080`. +The service listens on `http://localhost:8080`. To have something to +recall, load a set of sample datapoints and list them: + +```sh +cargo run --example seed # or: cargo run --example seed -- http://host:8080 +curl 'localhost:8080/api/v1/datapoints?projectId=0192f0c0-5eed-7000-8000-000000000001' +``` + +The seed uses `sourceId`s, so running it again updates the same +datapoints instead of adding copies. ### Docker @@ -239,6 +253,35 @@ with the `id`, the `sourceId` when given, and `deleted`. `deleted` is Deleting a datapoint that was stored with an `idempotencyKey` and then retrying that same request stores it again. +### `GET /api/v1/datapoints` + +``` +GET /api/v1/datapoints?projectId=0192...uuid&audience=sales&label=call&limit=20 +``` + +Lists a project's stored datapoints, read straight from S3, to check what +is in there. It also works while the index is still loading. + +| Parameter | Notes | +|------------------|--------------------------------------------------------------------| +| `projectId` | required | +| `limit` | 1 to 200, default 50 | +| `cursor` | `nextCursor` from the previous page | +| `sourceId` | only the datapoint with this `sourceId` | +| `audience` | only this audience; `project` means datapoints with none | +| `label` | only labels containing this text, ignoring case | +| `occurredFrom` | inclusive start on `timestamp` | +| `occurredTo` | inclusive end; start after end is `400 recall-range-invalid` | +| `includeContent` | `true` to include `content`; off by default since it costs an extra read each | + +Returns `{"datapoints": [...], "nextCursor": "..."}`. Each datapoint has +the same fields as in ingest, without `content` unless asked for, plus +`contentTokens`, `contentSha256`, `embeddingModel` and `embeddingDim`. +`nextCursor` is `null` on the last page. Pages are in `id` order, not by +time. One request reads at most 1000 datapoints, so with narrow filters a +page can have fewer than `limit` results, or none, and still a +`nextCursor`; keep going until it is `null`. + ### Webhooks A webhook turns events that another service POSTs (Jira, GitHub, an alert @@ -362,14 +405,16 @@ document that does not fit the remaining budget is skipped. ### `GET /healthz` Always `200`. The body is `{"status": "ok"}` once the search index has -been rebuilt from S3 and `{"status": "loading"}` before that. Ingest and -recall return `503` with code `loading` until then. +been rebuilt from S3 and `{"status": "loading"}` before that. Ingest, +recall, delete and webhook events return `503` with code `loading` until +then. Listing datapoints and managing webhooks work meanwhile. ## Not yet - Auth -- Filtering on `audience` -- Metadata filters on recall +- Filtering on `audience` in recall +- Metadata filters on recall and listing +- Signature checks on webhook events ## License diff --git a/src/api/datapoints.rs b/src/api/datapoints.rs new file mode 100644 index 0000000..a978ee3 --- /dev/null +++ b/src/api/datapoints.rs @@ -0,0 +1,252 @@ +use axum::extract::rejection::QueryRejection; +use axum::extract::{Query, State}; +use chrono::{DateTime, Utc}; +use futures::{StreamExt, TryStreamExt}; +use serde::{Deserialize, Serialize}; +use serde_json::{Map, Value}; +use uuid::Uuid; + +use crate::api::ApiError; +use crate::api::state::AppState; +use crate::domain::{Meta, PROJECT_AUDIENCE, source_uuid, validate_audience}; + +pub const DEFAULT_LIMIT: usize = 50; +pub const MAX_LIMIT: usize = 200; +/// Most datapoints read per request. With narrow filters a page can come +/// back short, with a cursor to carry on from. +pub const MAX_SCAN: usize = 1000; +/// Datapoints read per round when filtering. +const CHUNK: usize = 100; +const FETCH_CONCURRENCY: usize = 16; + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ListQuery { + pub project_id: Uuid, + pub limit: Option, + pub cursor: Option, + pub source_id: Option, + pub audience: Option, + pub label: Option, + pub occurred_from: Option>, + pub occurred_to: Option>, + #[serde(default)] + pub include_content: bool, +} + +/// A stored datapoint as listed: its metadata without the vector, and its +/// content only when asked for. +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct DatapointView { + pub id: Uuid, + pub project_id: Uuid, + pub source_id: Option, + pub label: String, + pub timestamp: DateTime, + pub metadata: Map, + pub audience: Option, + pub content_tokens: usize, + pub content_sha256: Option, + pub embedding_model: String, + pub embedding_dim: usize, + #[serde(skip_serializing_if = "Option::is_none")] + pub content: Option, +} + +impl DatapointView { + fn new(meta: Meta, content: Option) -> DatapointView { + DatapointView { + id: meta.id, + project_id: meta.project_id, + source_id: meta.source_id, + label: meta.label, + timestamp: meta.timestamp, + metadata: meta.metadata, + audience: meta.audience, + content_tokens: meta.content_tokens, + content_sha256: meta.content_sha256, + embedding_model: meta.embedding.model, + embedding_dim: meta.embedding.dim, + content, + } + } +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct DatapointList { + pub datapoints: Vec, + /// Pass as `cursor` for the next page; `null` on the last one. + pub next_cursor: Option, +} + +/// Which stored datapoints to list. +#[derive(Debug, Default)] +struct Filter { + /// `Some(None)` is the whole-project audience. + audience: Option>, + /// Lowercased. + label: Option, + from: Option>, + to: Option>, +} + +impl Filter { + fn from_query(q: &ListQuery) -> Result { + if let Some(audience) = &q.audience { + validate_audience(audience).map_err(ApiError::Validation)?; + } + if q.label.as_ref().is_some_and(|l| l.trim().is_empty()) { + return Err(ApiError::Validation( + "label must not be empty; omit it instead".into(), + )); + } + if let (Some(from), Some(to)) = (q.occurred_from, q.occurred_to) + && from > to + { + return Err(ApiError::RecallRangeInvalid); + } + Ok(Filter { + audience: q + .audience + .clone() + .map(|a| (a != PROJECT_AUDIENCE).then_some(a)), + label: q.label.as_ref().map(|l| l.to_lowercase()), + from: q.occurred_from, + to: q.occurred_to, + }) + } + + fn is_empty(&self) -> bool { + self.audience.is_none() && self.label.is_none() && self.from.is_none() && self.to.is_none() + } + + fn matches(&self, meta: &Meta) -> bool { + self.audience.as_ref().is_none_or(|a| *a == meta.audience) + && self + .label + .as_ref() + .is_none_or(|l| meta.label.to_lowercase().contains(l)) + && self.from.is_none_or(|from| meta.timestamp >= from) + && self.to.is_none_or(|to| meta.timestamp <= to) + } +} + +/// List a project's stored datapoints, read straight from storage, in id +/// order. Meant for debugging: it works while the index is still loading. +pub async fn list( + State(state): State, + query: Result, QueryRejection>, +) -> Result, ApiError> { + let Query(q) = query.map_err(|e| ApiError::Validation(e.body_text()))?; + let limit = q.limit.unwrap_or(DEFAULT_LIMIT); + if !(1..=MAX_LIMIT).contains(&limit) { + return Err(ApiError::Validation(format!( + "limit must be between 1 and {MAX_LIMIT}" + ))); + } + if q.source_id.as_ref().is_some_and(|s| s.trim().is_empty()) { + return Err(ApiError::Validation( + "sourceId must not be empty; omit it instead".into(), + )); + } + let cursor = q + .cursor + .as_deref() + .map(|c| { + c.parse::() + .map_err(|_| ApiError::Validation(format!("cursor {c:?} is not valid"))) + }) + .transpose()?; + let filter = Filter::from_query(&q)?; + let project = q.project_id; + + let (metas, next) = match &q.source_id { + Some(source_id) => { + let meta = state + .store + .get_meta(project, source_uuid(project, source_id)) + .await?; + ( + meta.into_iter().filter(|m| filter.matches(m)).collect(), + None, + ) + } + None => scan(&state, project, cursor, limit, &filter).await?, + }; + + let include_content = q.include_content; + let datapoints = futures::stream::iter(metas) + .map(|meta| { + let state = &state; + async move { + let content = if include_content { + state.store.get_content(project, meta.id).await? + } else { + None + }; + anyhow::Ok(DatapointView::new(meta, content)) + } + }) + .buffered(FETCH_CONCURRENCY) + .try_collect() + .await?; + Ok(axum::Json(DatapointList { + datapoints, + next_cursor: next.map(|id| id.to_string()), + })) +} + +/// Up to `limit` matching datapoints after `after`, reading at most +/// [`MAX_SCAN`], and the id to continue after when there may be more. +async fn scan( + state: &AppState, + project: Uuid, + mut after: Option, + limit: usize, + filter: &Filter, +) -> anyhow::Result<(Vec, Option)> { + let mut found = Vec::new(); + let mut scanned = 0; + loop { + // Without a filter every datapoint read is returned, so read no more + // than the page needs. + let want = if filter.is_empty() { + limit - found.len() + } else { + CHUNK + }; + let chunk = want.min(MAX_SCAN - scanned); + let ids = state + .store + .list_datapoint_ids(project, after, chunk) + .await?; + let exhausted = ids.len() < chunk; + let metas: Vec> = futures::stream::iter(ids.iter().copied()) + .map(|id| state.store.get_meta(project, id)) + .buffered(FETCH_CONCURRENCY) + .try_collect() + .await?; + let last = ids.last().copied(); + for (id, meta) in ids.into_iter().zip(metas) { + after = Some(id); + scanned += 1; + // Gone since listing, or filtered out. + let Some(meta) = meta.filter(|m| filter.matches(m)) else { + continue; + }; + found.push(meta); + if found.len() == limit { + let more = !(exhausted && Some(id) == last); + return Ok((found, more.then_some(id))); + } + } + if exhausted { + return Ok((found, None)); + } + if scanned >= MAX_SCAN { + return Ok((found, after)); + } + } +} diff --git a/src/api/mod.rs b/src/api/mod.rs index 3580b91..aaefa25 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -1,3 +1,4 @@ +pub mod datapoints; pub mod delete; pub mod error; pub mod ingest; @@ -39,6 +40,7 @@ pub fn router(state: AppState) -> Router { .route("/api/v1/ingest", post(ingest::handler)) .route("/api/v1/ingest/batch", post(ingest::batch_handler)) .route("/api/v1/recall", post(recall::handler)) + .route("/api/v1/datapoints", get(datapoints::list)) .route("/api/v1/delete", post(delete::handler)) .route( "/api/v1/webhooks", diff --git a/src/store.rs b/src/store.rs index f263940..4f34f12 100644 --- a/src/store.rs +++ b/src/store.rs @@ -371,6 +371,50 @@ impl S3Store { Ok(keys) } + /// Up to `max` of a project's datapoint ids in key order, starting after + /// `after`. Fewer than `max` means there are no more. + pub async fn list_datapoint_ids( + &self, + project: Uuid, + after: Option, + max: usize, + ) -> anyhow::Result> { + let prefix = format!("projects/{project}/datapoints/"); + // `meta.json` sorts after `content`, so this skips both of `after`'s keys. + let start_after = after.map(|id| keys::meta(project, id)); + // Two objects per datapoint. + let page_size = (max * 2 + 1).min(1000) as i32; + let mut pages = self + .client + .list_objects_v2() + .bucket(&self.bucket) + .prefix(&prefix) + .set_start_after(start_after) + .max_keys(page_size) + .into_paginator() + .send(); + let mut ids = Vec::new(); + while ids.len() < max + && let Some(page) = pages.next().await + { + let page = page.map_err(|e| { + explain( + e, + format!("listing {prefix} in {}", self.bucket), + &s3_permissions(&self.bucket), + ) + })?; + ids.extend( + page.contents() + .iter() + .filter_map(|o| keys::parse_meta_key(o.key()?)) + .map(|(_, id)| id), + ); + } + ids.truncate(max); + Ok(ids) + } + /// Every datapoint's metadata across all projects. pub async fn list_metas(&self) -> anyhow::Result> { let found: Vec<(Uuid, Uuid)> = self diff --git a/tests/datapoints.rs b/tests/datapoints.rs new file mode 100644 index 0000000..d42d899 --- /dev/null +++ b/tests/datapoints.rs @@ -0,0 +1,178 @@ +mod common; + +use axum::http::StatusCode; +use serde_json::{Value, json}; + +async fn ingest(app: &axum::Router, body: Value) -> Value { + let (status, resp) = common::post(app, "/api/v1/ingest", body).await; + assert_eq!(status, StatusCode::CREATED, "{resp}"); + resp +} + +fn labels(body: &Value) -> Vec { + let mut labels: Vec = body["datapoints"] + .as_array() + .unwrap() + .iter() + .map(|d| d["label"].as_str().unwrap().to_string()) + .collect(); + labels.sort(); + labels +} + +#[tokio::test] +#[ignore] +async fn lists_a_project_page_by_page() { + let (app, _) = common::app().await; + let (p, other) = (common::project(), common::project()); + for i in 0..5 { + ingest( + &app, + json!({ "projectId": p, "label": format!("dp{i}"), "content": format!("content {i}") }), + ) + .await; + } + ingest( + &app, + json!({ "projectId": other, "label": "elsewhere", "content": "x" }), + ) + .await; + + let mut seen = Vec::new(); + let mut cursor: Option = None; + let mut pages = 0; + loop { + let mut path = format!("/api/v1/datapoints?projectId={p}&limit=2"); + if let Some(c) = &cursor { + path.push_str(&format!("&cursor={c}")); + } + let (status, body) = common::get(&app, &path).await; + assert_eq!(status, StatusCode::OK, "{body}"); + pages += 1; + for d in body["datapoints"].as_array().unwrap() { + assert!(d.get("content").is_none(), "content only on request"); + assert!(d["contentTokens"].as_u64().unwrap() > 0); + assert_eq!(d["embeddingModel"], "fake"); + seen.push(d["label"].as_str().unwrap().to_string()); + } + match body["nextCursor"].as_str() { + Some(c) => cursor = Some(c.to_string()), + None => break, + } + } + assert_eq!(pages, 3); + seen.sort(); + assert_eq!(seen, ["dp0", "dp1", "dp2", "dp3", "dp4"]); +} + +#[tokio::test] +#[ignore] +async fn filters() { + let (app, _) = common::app().await; + let p = common::project(); + ingest( + &app, + json!({ "projectId": p, "sourceId": "crm-1", "label": "Kickoff call", "content": "agenda", + "timestamp": "2026-01-01T00:00:00Z", "audience": "sales" }), + ) + .await; + ingest( + &app, + json!({ "projectId": p, "label": "Infra decision", "content": "s3 vectors", + "timestamp": "2026-06-01T00:00:00Z" }), + ) + .await; + ingest( + &app, + json!({ "projectId": p, "label": "Follow-up call", "content": "notes", + "timestamp": "2026-09-01T00:00:00Z", "audience": "sales" }), + ) + .await; + + let list = |query: &str| { + let app = app.clone(); + let path = format!("/api/v1/datapoints?projectId={p}&{query}"); + async move { + let (status, body) = common::get(&app, &path).await; + assert_eq!(status, StatusCode::OK, "{body}"); + body + } + }; + + assert_eq!( + labels(&list("audience=sales").await), + ["Follow-up call", "Kickoff call"] + ); + assert_eq!(labels(&list("audience=project").await), ["Infra decision"]); + assert_eq!( + labels(&list("label=CALL").await), + ["Follow-up call", "Kickoff call"] + ); + assert_eq!( + labels(&list("occurredFrom=2026-05-01T00:00:00Z&occurredTo=2026-07-01T00:00:00Z").await), + ["Infra decision"] + ); + assert_eq!( + labels(&list("label=call&occurredFrom=2026-08-01T00:00:00Z").await), + ["Follow-up call"] + ); + + let body = list("sourceId=crm-1&includeContent=true").await; + assert_eq!(body["datapoints"][0]["content"], "agenda"); + assert_eq!(body["datapoints"][0]["sourceId"], "crm-1"); + assert!(body["nextCursor"].is_null()); + assert_eq!( + labels(&list("sourceId=crm-1&audience=project").await).len(), + 0 + ); + assert_eq!(labels(&list("sourceId=never-stored").await).len(), 0); +} + +#[tokio::test] +#[ignore] +async fn works_while_loading() { + let (app, state) = common::app().await; + let p = common::project(); + ingest( + &app, + json!({ "projectId": p, "label": "a", "content": "b" }), + ) + .await; + state + .ready + .store(false, std::sync::atomic::Ordering::SeqCst); + let (status, body) = common::get(&app, &format!("/api/v1/datapoints?projectId={p}")).await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(labels(&body), ["a"]); +} + +#[tokio::test] +#[ignore] +async fn list_validation() { + let (app, _) = common::app().await; + let p = common::project(); + for query in [ + String::new(), + "projectId=nope".into(), + format!("projectId={p}&limit=0"), + format!("projectId={p}&limit=201"), + format!("projectId={p}&cursor=nope"), + format!("projectId={p}&audience=Sales"), + format!("projectId={p}&label=%20"), + format!("projectId={p}&sourceId="), + format!("projectId={p}&occurredFrom=yesterday"), + ] { + let (status, body) = common::get(&app, &format!("/api/v1/datapoints?{query}")).await; + assert_eq!(status, StatusCode::BAD_REQUEST, "{query}: {body}"); + assert_eq!(body["error"]["code"], "validation", "{query}"); + } + let (status, body) = common::get( + &app, + &format!( + "/api/v1/datapoints?projectId={p}&occurredFrom=2026-02-01T00:00:00Z&occurredTo=2026-01-01T00:00:00Z" + ), + ) + .await; + assert_eq!(status, StatusCode::BAD_REQUEST); + assert_eq!(body["error"]["code"], "recall-range-invalid"); +}