chore: scaffold moments workspace

Cargo workspace with five crates per architecture conventions:

- moments-entities: Source enum, Event, EventQuery, SourceSummary
- moments-core:     EventReader / EventWriter ports
- moments-data:     PgStore (sqlx postgres adapter) + 0001_init.sql
- moments-api:      axum binary; /v1/{healthz,events,sources}
- moments-worker:   skeleton; pollers land in step 2

Sources committed-to for ingestion: github, gitea, hg, bugzilla.
Workstation events explicitly retired (not deferred).

Build + clippy clean. sqlx queries use the runtime API for now;
will switch to compile-time-checked macros + .sqlx offline cache
once magrathea has the moments_{ro,rw} roles and database created.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

crates/moments-api/Cargo.toml

@@ -0,0 +1,22 @@
+[package]
+name = "moments-api"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+authors.workspace = true
+
+[dependencies]
+moments-entities.workspace = true
+moments-core.workspace = true
+moments-data.workspace = true
+tokio.workspace = true
+axum.workspace = true
+tower-http.workspace = true
+tracing.workspace = true
+tracing-subscriber.workspace = true
+anyhow.workspace = true
+serde.workspace = true
+serde_json.workspace = true
+chrono.workspace = true
+clap.workspace = true

crates/moments-api/src/main.rs

@@ -0,0 +1,146 @@
+use std::{net::SocketAddr, sync::Arc};
+
+use axum::{
+    Json, Router,
+    extract::{Query, State},
+    http::StatusCode,
+    response::IntoResponse,
+    routing::get,
+};
+use chrono::{DateTime, Utc};
+use clap::Parser;
+use moments_core::EventReader;
+use moments_data::PgStore;
+use moments_entities::{Event, EventQuery, Source, SourceSummary};
+use serde::Deserialize;
+use tower_http::{cors::CorsLayer, trace::TraceLayer};
+use tracing::info;
+
+#[derive(Parser, Debug)]
+#[command(version, about = "moments read-only HTTP API")]
+struct Args {
+    #[arg(long, env = "BIND_ADDR", default_value = "127.0.0.1:8080")]
+    bind: SocketAddr,
+
+    #[arg(long, env = "DATABASE_URL")]
+    database_url: String,
+}
+
+#[derive(Clone)]
+struct AppState {
+    store: Arc<PgStore>,
+}
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    init_tracing();
+    let args = Args::parse();
+
+    let store = PgStore::connect(&args.database_url).await?;
+    store.migrate().await?;
+    let state = AppState {
+        store: Arc::new(store),
+    };
+
+    let app = Router::new()
+        .route("/v1/healthz", get(healthz))
+        .route("/v1/events", get(list_events))
+        .route("/v1/sources", get(list_sources))
+        .with_state(state)
+        .layer(TraceLayer::new_for_http())
+        .layer(CorsLayer::permissive());
+
+    info!(addr = %args.bind, "listening");
+    let listener = tokio::net::TcpListener::bind(args.bind).await?;
+    axum::serve(listener, app).await?;
+    Ok(())
+}
+
+fn init_tracing() {
+    use tracing_subscriber::{EnvFilter, fmt};
+    let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
+    let json = std::env::var("JOURNAL_STREAM").is_ok();
+    if json {
+        fmt().with_env_filter(filter).json().init();
+    } else {
+        fmt().with_env_filter(filter).init();
+    }
+}
+
+async fn healthz() -> &'static str {
+    "ok"
+}
+
+#[derive(Debug, Deserialize)]
+struct EventsQueryParams {
+    from: Option<DateTime<Utc>>,
+    to: Option<DateTime<Utc>>,
+    /// Comma-separated list, e.g. `source=github,gitea`.
+    source: Option<String>,
+    limit: Option<u32>,
+}
+
+async fn list_events(
+    State(state): State<AppState>,
+    Query(params): Query<EventsQueryParams>,
+) -> Result<Json<Vec<Event>>, ApiError> {
+    let sources = params
+        .source
+        .as_deref()
+        .map(parse_sources)
+        .transpose()?;
+
+    let limit = params.limit.unwrap_or(100).clamp(1, 1000);
+
+    let query = EventQuery {
+        from: params.from,
+        to: params.to,
+        sources,
+        limit,
+    };
+
+    let events = state.store.list_events(&query).await.map_err(internal)?;
+    Ok(Json(events))
+}
+
+async fn list_sources(
+    State(state): State<AppState>,
+) -> Result<Json<Vec<SourceSummary>>, ApiError> {
+    let summaries = state.store.source_summaries().await.map_err(internal)?;
+    Ok(Json(summaries))
+}
+
+fn parse_sources(raw: &str) -> Result<Vec<Source>, ApiError> {
+    raw.split(',')
+        .map(str::trim)
+        .filter(|s| !s.is_empty())
+        .map(|s| s.parse::<Source>().map_err(|e| ApiError::bad_request(e.to_string())))
+        .collect()
+}
+
+struct ApiError {
+    status: StatusCode,
+    message: String,
+}
+
+impl ApiError {
+    fn bad_request(msg: impl Into<String>) -> Self {
+        Self {
+            status: StatusCode::BAD_REQUEST,
+            message: msg.into(),
+        }
+    }
+}
+
+fn internal<E: std::fmt::Display>(e: E) -> ApiError {
+    ApiError {
+        status: StatusCode::INTERNAL_SERVER_ERROR,
+        message: e.to_string(),
+    }
+}
+
+impl IntoResponse for ApiError {
+    fn into_response(self) -> axum::response::Response {
+        (self.status, Json(serde_json::json!({ "error": self.message }))).into_response()
+    }
+}

crates/moments-core/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "moments-core"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+authors.workspace = true
+
+[dependencies]
+moments-entities.workspace = true
+serde.workspace = true
+serde_json.workspace = true
+chrono.workspace = true
+thiserror.workspace = true
+async-trait = "0.1"

crates/moments-core/src/lib.rs

@@ -0,0 +1,21 @@
+use async_trait::async_trait;
+use moments_entities::{Event, EventQuery, SourceSummary};
+
+#[derive(Debug, thiserror::Error)]
+pub enum StoreError {
+    #[error("database error: {0}")]
+    Database(String),
+}
+
+/// Read-side port consumed by `moments-api`.
+#[async_trait]
+pub trait EventReader: Send + Sync {
+    async fn list_events(&self, query: &EventQuery) -> Result<Vec<Event>, StoreError>;
+    async fn source_summaries(&self) -> Result<Vec<SourceSummary>, StoreError>;
+}
+
+/// Write-side port consumed by `moments-worker`. Idempotent upserts on `id`.
+#[async_trait]
+pub trait EventWriter: Send + Sync {
+    async fn upsert_events(&self, events: &[Event]) -> Result<usize, StoreError>;
+}

crates/moments-data/Cargo.toml

@@ -0,0 +1,17 @@
+[package]
+name = "moments-data"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+authors.workspace = true
+
+[dependencies]
+moments-entities.workspace = true
+moments-core.workspace = true
+sqlx.workspace = true
+serde_json.workspace = true
+chrono.workspace = true
+thiserror.workspace = true
+tracing.workspace = true
+async-trait = "0.1"

crates/moments-data/migrations/0001_init.sql

@@ -0,0 +1,11 @@
+CREATE TABLE events (
+    id          TEXT        PRIMARY KEY,
+    source      TEXT        NOT NULL,
+    action      TEXT        NOT NULL,
+    occurred_at TIMESTAMPTZ NOT NULL,
+    payload     JSONB       NOT NULL,
+    fetched_at  TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX events_occurred_at_desc        ON events (occurred_at DESC);
+CREATE INDEX events_source_occurred_at_desc ON events (source, occurred_at DESC);

crates/moments-data/src/lib.rs

@@ -0,0 +1,143 @@
+use async_trait::async_trait;
+use chrono::{DateTime, Utc};
+use moments_core::{EventReader, EventWriter, StoreError};
+use moments_entities::{Event, EventQuery, Source, SourceSummary};
+use sqlx::Row;
+use sqlx::postgres::{PgPool, PgPoolOptions};
+use std::str::FromStr;
+
+pub static MIGRATOR: sqlx::migrate::Migrator = sqlx::migrate!("./migrations");
+
+#[derive(Clone)]
+pub struct PgStore {
+    pool: PgPool,
+}
+
+impl PgStore {
+    pub async fn connect(database_url: &str) -> Result<Self, sqlx::Error> {
+        let pool = PgPoolOptions::new()
+            .max_connections(8)
+            .connect(database_url)
+            .await?;
+        Ok(Self { pool })
+    }
+
+    pub async fn migrate(&self) -> Result<(), sqlx::migrate::MigrateError> {
+        MIGRATOR.run(&self.pool).await
+    }
+}
+
+fn map_err<E: std::fmt::Display>(e: E) -> StoreError {
+    StoreError::Database(e.to_string())
+}
+
+#[async_trait]
+impl EventReader for PgStore {
+    async fn list_events(&self, query: &EventQuery) -> Result<Vec<Event>, StoreError> {
+        let sources: Option<Vec<String>> = query
+            .sources
+            .as_ref()
+            .map(|s| s.iter().map(|x| x.as_str().to_string()).collect());
+
+        let rows = sqlx::query(
+            r#"
+            SELECT id, source, action, occurred_at, payload
+            FROM events
+            WHERE ($1::timestamptz IS NULL OR occurred_at >= $1)
+              AND ($2::timestamptz IS NULL OR occurred_at <  $2)
+              AND ($3::text[]      IS NULL OR source = ANY($3))
+            ORDER BY occurred_at DESC
+            LIMIT $4
+            "#,
+        )
+        .bind(query.from)
+        .bind(query.to)
+        .bind(sources.as_deref())
+        .bind(query.limit as i64)
+        .fetch_all(&self.pool)
+        .await
+        .map_err(map_err)?;
+
+        rows.into_iter()
+            .map(|r| {
+                let source_str: String = r.try_get("source").map_err(map_err)?;
+                Ok(Event {
+                    id: r.try_get("id").map_err(map_err)?,
+                    source: Source::from_str(&source_str).map_err(map_err)?,
+                    action: r.try_get("action").map_err(map_err)?,
+                    occurred_at: r.try_get("occurred_at").map_err(map_err)?,
+                    payload: r.try_get("payload").map_err(map_err)?,
+                })
+            })
+            .collect()
+    }
+
+    async fn source_summaries(&self) -> Result<Vec<SourceSummary>, StoreError> {
+        let rows = sqlx::query(
+            r#"
+            SELECT source,
+                   COUNT(*)         AS count,
+                   MIN(occurred_at) AS earliest,
+                   MAX(occurred_at) AS latest
+            FROM events
+            GROUP BY source
+            ORDER BY source
+            "#,
+        )
+        .fetch_all(&self.pool)
+        .await
+        .map_err(map_err)?;
+
+        rows.into_iter()
+            .map(|r| {
+                let source_str: String = r.try_get("source").map_err(map_err)?;
+                let earliest: Option<DateTime<Utc>> = r.try_get("earliest").map_err(map_err)?;
+                let latest: Option<DateTime<Utc>> = r.try_get("latest").map_err(map_err)?;
+                let count: i64 = r.try_get("count").map_err(map_err)?;
+                Ok(SourceSummary {
+                    source: Source::from_str(&source_str).map_err(map_err)?,
+                    count,
+                    earliest,
+                    latest,
+                })
+            })
+            .collect()
+    }
+}
+
+#[async_trait]
+impl EventWriter for PgStore {
+    async fn upsert_events(&self, events: &[Event]) -> Result<usize, StoreError> {
+        if events.is_empty() {
+            return Ok(0);
+        }
+
+        let mut tx = self.pool.begin().await.map_err(map_err)?;
+        let mut inserted = 0;
+        for ev in events {
+            let n = sqlx::query(
+                r#"
+                INSERT INTO events (id, source, action, occurred_at, payload)
+                VALUES ($1, $2, $3, $4, $5)
+                ON CONFLICT (id) DO UPDATE
+                SET source      = EXCLUDED.source,
+                    action      = EXCLUDED.action,
+                    occurred_at = EXCLUDED.occurred_at,
+                    payload     = EXCLUDED.payload
+                "#,
+            )
+            .bind(&ev.id)
+            .bind(ev.source.as_str())
+            .bind(&ev.action)
+            .bind(ev.occurred_at)
+            .bind(&ev.payload)
+            .execute(&mut *tx)
+            .await
+            .map_err(map_err)?
+            .rows_affected();
+            inserted += n as usize;
+        }
+        tx.commit().await.map_err(map_err)?;
+        Ok(inserted)
+    }
+}

crates/moments-entities/Cargo.toml

@@ -0,0 +1,13 @@
+[package]
+name = "moments-entities"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+authors.workspace = true
+
+[dependencies]
+serde.workspace = true
+serde_json.workspace = true
+chrono.workspace = true
+thiserror.workspace = true

crates/moments-entities/src/lib.rs

@@ -0,0 +1,76 @@
+use chrono::{DateTime, Utc};
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Hash)]
+#[serde(rename_all = "lowercase")]
+pub enum Source {
+    Github,
+    Gitea,
+    Hg,
+    Bugzilla,
+}
+
+impl Source {
+    pub const ALL: &'static [Source] = &[
+        Source::Github,
+        Source::Gitea,
+        Source::Hg,
+        Source::Bugzilla,
+    ];
+
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            Source::Github => "github",
+            Source::Gitea => "gitea",
+            Source::Hg => "hg",
+            Source::Bugzilla => "bugzilla",
+        }
+    }
+}
+
+impl std::str::FromStr for Source {
+    type Err = ParseSourceError;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s {
+            "github" => Ok(Source::Github),
+            "gitea" => Ok(Source::Gitea),
+            "hg" => Ok(Source::Hg),
+            "bugzilla" => Ok(Source::Bugzilla),
+            other => Err(ParseSourceError(other.to_string())),
+        }
+    }
+}
+
+#[derive(Debug, thiserror::Error)]
+#[error("unknown source: {0}")]
+pub struct ParseSourceError(pub String);
+
+/// Raw event as stored. The presentation reshape lives in `moments-core`
+/// and runs at API request time.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct Event {
+    pub id: String,
+    pub source: Source,
+    pub action: String,
+    pub occurred_at: DateTime<Utc>,
+    pub payload: serde_json::Value,
+}
+
+/// Filters accepted by `GET /v1/events`.
+#[derive(Debug, Clone, Default)]
+pub struct EventQuery {
+    pub from: Option<DateTime<Utc>>,
+    pub to: Option<DateTime<Utc>>,
+    pub sources: Option<Vec<Source>>,
+    pub limit: u32,
+}
+
+/// Per-source rollup returned by `GET /v1/sources`.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct SourceSummary {
+    pub source: Source,
+    pub count: i64,
+    pub earliest: Option<DateTime<Utc>>,
+    pub latest: Option<DateTime<Utc>>,
+}

crates/moments-worker/Cargo.toml

@@ -0,0 +1,21 @@
+[package]
+name = "moments-worker"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+authors.workspace = true
+
+[dependencies]
+moments-entities.workspace = true
+moments-core.workspace = true
+moments-data.workspace = true
+tokio.workspace = true
+tracing.workspace = true
+tracing-subscriber.workspace = true
+anyhow.workspace = true
+clap.workspace = true
+reqwest.workspace = true
+serde.workspace = true
+serde_json.workspace = true
+chrono.workspace = true

crates/moments-worker/src/main.rs

@@ -0,0 +1,37 @@
+use clap::Parser;
+use moments_data::PgStore;
+use tracing::info;
+
+#[derive(Parser, Debug)]
+#[command(version, about = "moments ingestion worker")]
+struct Args {
+    #[arg(long, env = "DATABASE_URL")]
+    database_url: String,
+}
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    init_tracing();
+    let args = Args::parse();
+
+    let store = PgStore::connect(&args.database_url).await?;
+    store.migrate().await?;
+
+    info!("worker started — pollers will land in step 2");
+
+    // Pollers (github, gitea, hg, bugzilla) land in subsequent steps.
+    // For now this binary only verifies it can reach the database.
+    let _ = store;
+    Ok(())
+}
+
+fn init_tracing() {
+    use tracing_subscriber::{EnvFilter, fmt};
+    let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
+    let json = std::env::var("JOURNAL_STREAM").is_ok();
+    if json {
+        fmt().with_env_filter(filter).json().init();
+    } else {
+        fmt().with_env_filter(filter).init();
+    }
+}