feat(world): instant news via warm feed cache + queryable ingested-data lake
Make news/feeds INSTANT and give world one place to query everything. Feed hot cache (two-tier, ONE way to read a feed body): - L1 per-pod in-mem mirror (sub-ms hot reads) + L2 hanzo-kv (shared across pods, survives restart). A body warmed by any pod is instantly served by every pod; a restarted pod reads the still-warm shared cache instead of cold-starting. Graceful degrade to L1-only when hanzo-kv is unreachable. - Background warmer: on boot + every ~5m (jittered) fetches every warm feed, write-throughs the body, and folds items into the lake. rss-proxy and feeds-batch serve from the warm cache and never block on upstream while any cached copy exists (stale-while-revalidate; revalidation runs in background). - Demand-driven warm set (fleet-wide via hanzo-kv) + a curated seed of the crypto / financial-regulation / crypto-news feeds so a brand-new pod warms THOSE first. Cross-pod fetch dedupe via shared copy freshness. Ingested-data lake (embedded SQLite, modernc.org/sqlite — pure Go, FTS5): - One normalized `items` table (id, kind, source, ts, title, text, tickers, country, geo, payload) fed by feed news AND world-model observations, so everything is searchable/countable together. - GET /v1/world/search (FTS5 bm25 ranking; filters kind/since/country/ticker) and GET /v1/world/analytics (totals, by kind/source, top tickers). Rolling retention window with an hourly prune job. Write-behind ingest (never on the request path). Per-identity settings (same SQLite DB, own table): - GET/PUT /v1/world/settings, bearer-gated, keyed by (org, user_sub, project) from IAM userinfo, so a signed-in dashboard syncs across devices. Anonymous → 401 (client keeps localStorage). Everything degrades cleanly (never-5xx): no hanzo-kv → per-pod cache; no SQLite → empty search/analytics + settings say not-stored. Binary stays CGO-free (CGO_ENABLED=0). go directive + Dockerfile bumped to 1.25 for modernc. Latency (local, warm): feeds-batch warm 0.5-1.9ms vs 154ms cold; search 0.8-1.5ms. Claude-Session: https://claude.ai/code/session_013jh8aka8q8RvhhVQ1psMeW
This commit is contained in:
hanzo-dev
+7
-3
@@ -37,10 +37,14 @@ ENV VITE_MAPBOX_TOKEN=$VITE_MAPBOX_TOKEN
|
||||
RUN npm run build
|
||||
|
||||
# ---- go stage: build the static server binary (CGO-free) -----------------
|
||||
FROM golang:1.23-alpine AS gobuild
|
||||
# go 1.25: the embedded datastore (modernc.org/sqlite, pure Go) needs it. The
|
||||
# binary stays CGO-free — modernc's SQLite is pure Go, so no C toolchain is added.
|
||||
FROM golang:1.25-alpine AS gobuild
|
||||
WORKDIR /src
|
||||
# stdlib-only module: go.mod has no requires, so there is no go.sum to copy.
|
||||
COPY go.mod ./
|
||||
# Deps: hanzo-kv client (go-redis) + embedded SQLite (modernc). Download once for
|
||||
# a cached layer before the source is copied.
|
||||
COPY go.mod go.sum ./
|
||||
RUN go mod download
|
||||
COPY cmd ./cmd
|
||||
COPY internal ./internal
|
||||
RUN CGO_ENABLED=0 GOOS=linux go build -trimpath -ldflags="-s -w" -o /out/world ./cmd/world
|
||||
|
||||
@@ -46,7 +46,9 @@ func main() {
|
||||
world.LoadKMSSecrets(rootCtx)
|
||||
|
||||
srv := world.NewServer()
|
||||
defer srv.Close() // release hanzo-kv + embedded datastore handles
|
||||
srv.StartModel(rootCtx) // continuously-folded world-state engine
|
||||
srv.StartDatastore(rootCtx) // shared feed warmer + lake write-behind/prune
|
||||
mux := http.NewServeMux()
|
||||
srv.Mount(mux) // /v1/world/* routes
|
||||
|
||||
|
||||
@@ -1,3 +1,24 @@
|
||||
module github.com/hanzoai/world
|
||||
|
||||
go 1.23
|
||||
go 1.25.0
|
||||
|
||||
require (
|
||||
github.com/alicebob/miniredis/v2 v2.38.0
|
||||
github.com/redis/go-redis/v9 v9.20.0
|
||||
modernc.org/sqlite v1.51.0
|
||||
)
|
||||
|
||||
require (
|
||||
github.com/cespare/xxhash/v2 v2.3.0 // indirect
|
||||
github.com/dustin/go-humanize v1.0.1 // indirect
|
||||
github.com/google/uuid v1.6.0 // indirect
|
||||
github.com/mattn/go-isatty v0.0.20 // indirect
|
||||
github.com/ncruces/go-strftime v1.0.0 // indirect
|
||||
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
|
||||
github.com/yuin/gopher-lua v1.1.1 // indirect
|
||||
go.uber.org/atomic v1.11.0 // indirect
|
||||
golang.org/x/sys v0.42.0 // indirect
|
||||
modernc.org/libc v1.72.3 // indirect
|
||||
modernc.org/mathutil v1.7.1 // indirect
|
||||
modernc.org/memory v1.11.0 // indirect
|
||||
)
|
||||
|
||||
@@ -0,0 +1,75 @@
|
||||
github.com/alicebob/miniredis/v2 v2.38.0 h1:nZAzCR+Lj+Vxk4ZXzm2NuKq2O33RXj1XxJ2e2uP9jiw=
|
||||
github.com/alicebob/miniredis/v2 v2.38.0/go.mod h1:TcL7YfarKPGDAthEtl5NBeHZfeUQj6OXMm/+iu5cLMM=
|
||||
github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
|
||||
github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
|
||||
github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
|
||||
github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
|
||||
github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
|
||||
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
|
||||
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
|
||||
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
|
||||
github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
|
||||
github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e h1:ijClszYn+mADRFY17kjQEVQ1XRhq2/JR1M3sGqeJoxs=
|
||||
github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e/go.mod h1:boTsfXsheKC2y+lKOCMpSfarhxDeIzfZG1jqGcPl3cA=
|
||||
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
|
||||
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
|
||||
github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=
|
||||
github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
|
||||
github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE=
|
||||
github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0=
|
||||
github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
|
||||
github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
|
||||
github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w=
|
||||
github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls=
|
||||
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
|
||||
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
github.com/redis/go-redis/v9 v9.20.0 h1:WnQYxLkgO2xiXTCJY0ldIiI8dNqCDlQAG+AtaH7a2a0=
|
||||
github.com/redis/go-redis/v9 v9.20.0/go.mod h1:v/M13XI1PVCDcm01VtPFOADfZtHf8YW3baQf57KlIkA=
|
||||
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE=
|
||||
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
|
||||
github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0Q=
|
||||
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
|
||||
github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M=
|
||||
github.com/yuin/gopher-lua v1.1.1/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7kdAd1Pw=
|
||||
github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs=
|
||||
github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s=
|
||||
go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE=
|
||||
go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
|
||||
golang.org/x/mod v0.33.0 h1:tHFzIWbBifEmbwtGz65eaWyGiGZatSrT9prnU8DbVL8=
|
||||
golang.org/x/mod v0.33.0/go.mod h1:swjeQEj+6r7fODbD2cqrnje9PnziFuw4bmLbBZFrQ5w=
|
||||
golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
|
||||
golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
|
||||
golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo=
|
||||
golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
|
||||
golang.org/x/tools v0.42.0 h1:uNgphsn75Tdz5Ji2q36v/nsFSfR/9BRFvqhGBaJGd5k=
|
||||
golang.org/x/tools v0.42.0/go.mod h1:Ma6lCIwGZvHK6XtgbswSoWroEkhugApmsXyrUmBhfr0=
|
||||
modernc.org/cc/v4 v4.28.2 h1:3tQ0lf2ADtoby2EtSP+J7IE2SHwEJdP8ioR59wx7XpY=
|
||||
modernc.org/cc/v4 v4.28.2/go.mod h1:OnovgIhbbMXMu1aISnJ0wvVD1KnW+cAUJkIrAWh+kVI=
|
||||
modernc.org/ccgo/v4 v4.34.0 h1:yRLPFZieg532OT4rp4JFNIVcquwalMX26G95WQDqwCQ=
|
||||
modernc.org/ccgo/v4 v4.34.0/go.mod h1:AS5WYMyBakQ+fhsHhtP8mWB82KTGPkNNJDGfGQCe0/A=
|
||||
modernc.org/fileutil v1.4.0 h1:j6ZzNTftVS054gi281TyLjHPp6CPHr2KCxEXjEbD6SM=
|
||||
modernc.org/fileutil v1.4.0/go.mod h1:EqdKFDxiByqxLk8ozOxObDSfcVOv/54xDs/DUHdvCUU=
|
||||
modernc.org/gc/v2 v2.6.5 h1:nyqdV8q46KvTpZlsw66kWqwXRHdjIlJOhG6kxiV/9xI=
|
||||
modernc.org/gc/v2 v2.6.5/go.mod h1:YgIahr1ypgfe7chRuJi2gD7DBQiKSLMPgBQe9oIiito=
|
||||
modernc.org/gc/v3 v3.1.2 h1:ZtDCnhonXSZexk/AYsegNRV1lJGgaNZJuKjJSWKyEqo=
|
||||
modernc.org/gc/v3 v3.1.2/go.mod h1:HFK/6AGESC7Ex+EZJhJ2Gni6cTaYpSMmU/cT9RmlfYY=
|
||||
modernc.org/goabi0 v0.2.0 h1:HvEowk7LxcPd0eq6mVOAEMai46V+i7Jrj13t4AzuNks=
|
||||
modernc.org/goabi0 v0.2.0/go.mod h1:CEFRnnJhKvWT1c1JTI3Avm+tgOWbkOu5oPA8eH8LnMI=
|
||||
modernc.org/libc v1.72.3 h1:ZnDF4tXn4NBXFutMMQC4vtbTFSXhhKzR73fv0beZEAU=
|
||||
modernc.org/libc v1.72.3/go.mod h1:dn0dZNnnn1clLyvRxLxYExxiKRZIRENOfqQ8XEeg4Qs=
|
||||
modernc.org/mathutil v1.7.1 h1:GCZVGXdaN8gTqB1Mf/usp1Y/hSqgI2vAGGP4jZMCxOU=
|
||||
modernc.org/mathutil v1.7.1/go.mod h1:4p5IwJITfppl0G4sUEDtCr4DthTaT47/N3aT6MhfgJg=
|
||||
modernc.org/memory v1.11.0 h1:o4QC8aMQzmcwCK3t3Ux/ZHmwFPzE6hf2Y5LbkRs+hbI=
|
||||
modernc.org/memory v1.11.0/go.mod h1:/JP4VbVC+K5sU2wZi9bHoq2MAkCnrt2r98UGeSK7Mjw=
|
||||
modernc.org/opt v0.2.0 h1:tGyef5ApycA7FSEOMraay9SaTk5zmbx7Tu+cJs4QKZg=
|
||||
modernc.org/opt v0.2.0/go.mod h1:03fq9lsNfvkYSfxrfUhZCWPk1lm4cq4N+Bh//bEtgns=
|
||||
modernc.org/sortutil v1.2.1 h1:+xyoGf15mM3NMlPDnFqrteY07klSFxLElE2PVuWIJ7w=
|
||||
modernc.org/sortutil v1.2.1/go.mod h1:7ZI3a3REbai7gzCLcotuw9AC4VZVpYMjDzETGsSMqJE=
|
||||
modernc.org/sqlite v1.51.0 h1:aH/MMSoayAIhozZ7uJbVTT9QO/VhzBf0J9tymmmuC/U=
|
||||
modernc.org/sqlite v1.51.0/go.mod h1:tcNzv5p84E0skkmJn038y+hWJbLQXQqEnQfeh5r2JLM=
|
||||
modernc.org/strutil v1.2.1 h1:UneZBkQA+DX2Rp35KcM69cSsNES9ly8mQWD71HKlOA0=
|
||||
modernc.org/strutil v1.2.1/go.mod h1:EHkiggD70koQxjVdSBM3JKM7k6L0FbGE5eymy9i3B9A=
|
||||
modernc.org/token v1.1.0 h1:Xl7Ap9dKaEs5kLoOQeQmPWevfnk/DM5qcLcYlA8ys6Y=
|
||||
modernc.org/token v1.1.0/go.mod h1:UGzOrNV1mAFSEB63lOFHIpNRUVMvYTc6yu1SMY/XTDM=
|
||||
@@ -0,0 +1,123 @@
|
||||
package world
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"time"
|
||||
|
||||
"github.com/hanzoai/world/internal/world/kv"
|
||||
"github.com/hanzoai/world/internal/world/model"
|
||||
"github.com/hanzoai/world/internal/world/store"
|
||||
)
|
||||
|
||||
// datastore.go wires world's three storage concerns onto the Server and owns
|
||||
// their lifecycle, keeping each in its lane:
|
||||
// - hanzo-kv (shared hot cache) → instant feed bodies (FeedCache L2)
|
||||
// - embedded SQLite (lake) → the searchable "one place to query everything"
|
||||
// - embedded SQLite (settings) → signed-in per-identity dashboard sync
|
||||
//
|
||||
// Everything degrades cleanly: no hanzo-kv → per-pod in-mem feed cache; no
|
||||
// SQLite → search/analytics return empty and settings say "not stored". The
|
||||
// service never 5xxes over storage.
|
||||
|
||||
// kvAddr is the hanzo-kv endpoint. Defaults to the in-cluster Service; set empty
|
||||
// (WORLD_KV_DISABLE=1) to force the pure in-mem path (local dev / CI).
|
||||
func kvAddr() string {
|
||||
if env("WORLD_KV_DISABLE") != "" {
|
||||
return ""
|
||||
}
|
||||
if a := env("HANZO_KV_ADDR", "WORLD_KV_ADDR"); a != "" {
|
||||
return a
|
||||
}
|
||||
return "hanzo-kv:6379"
|
||||
}
|
||||
|
||||
// kvPassword is optional — hanzo-kv currently requires none; the hook is here for
|
||||
// a future KMS-provisioned password (HANZO_KV_PASSWORD / world-secrets).
|
||||
func kvPassword() string { return env("HANZO_KV_PASSWORD", "WORLD_KV_PASSWORD") }
|
||||
|
||||
// lakeRetention is the rolling window ingested items are kept for.
|
||||
func lakeRetention() time.Duration {
|
||||
if v := env("WORLD_LAKE_RETENTION"); v != "" {
|
||||
if d, err := time.ParseDuration(v); err == nil && d > 0 {
|
||||
return d
|
||||
}
|
||||
}
|
||||
return store.DefaultRetention
|
||||
}
|
||||
|
||||
// initDatastore opens hanzo-kv + the embedded SQLite datastore and builds the
|
||||
// two-tier feed cache. Called once from NewServer; never fails hard.
|
||||
func (s *Server) initDatastore() {
|
||||
s.kv = kv.Open(kvAddr(), kvPassword())
|
||||
|
||||
db, err := store.Open(modelDataDir(), lakeRetention())
|
||||
if err != nil {
|
||||
logf("world-store: degraded (no durable lake/settings): %v", err)
|
||||
}
|
||||
s.store = db
|
||||
s.feeds = NewFeedCache(s.kv, 0, curatedFeedSeed)
|
||||
|
||||
// The world model dumps its folded observations into the lake so model state
|
||||
// is queryable alongside news — the engine stays decomplected from storage,
|
||||
// it just calls this sink.
|
||||
s.worldModel.SetObservationSink(s.ingestObservations)
|
||||
}
|
||||
|
||||
// StartDatastore starts the background loops: the lake write-behind/prune
|
||||
// consumer and the feed warmer. Call once from main after the server is built.
|
||||
func (s *Server) StartDatastore(ctx context.Context) {
|
||||
if s.kv.Enabled() {
|
||||
pctx, cancel := context.WithTimeout(ctx, 3*time.Second)
|
||||
if err := s.kv.Ping(pctx); err != nil {
|
||||
logf("world-kv: %s unreachable, using per-pod in-mem cache: %v", kvAddr(), err)
|
||||
} else {
|
||||
logf("world-kv: connected to %s (shared feed cache)", kvAddr())
|
||||
}
|
||||
cancel()
|
||||
} else {
|
||||
logf("world-kv: disabled, using per-pod in-mem feed cache")
|
||||
}
|
||||
go s.store.Lake.Run(ctx)
|
||||
s.startFeedWarmer(ctx)
|
||||
}
|
||||
|
||||
// Close releases the datastore handles. Safe to call once on shutdown.
|
||||
func (s *Server) Close() {
|
||||
if s.kv != nil {
|
||||
s.kv.Close()
|
||||
}
|
||||
if s.store != nil {
|
||||
_ = s.store.Close()
|
||||
}
|
||||
}
|
||||
|
||||
// ingestObservations folds one cycle of world-model observations into the lake
|
||||
// (kind=observation), keyed so each entity+source keeps its latest value. This
|
||||
// is the model half of "dump ALL ingested data into the datastore".
|
||||
func (s *Server) ingestObservations(obs []model.Observation) {
|
||||
if s.store == nil {
|
||||
return
|
||||
}
|
||||
now := time.Now().UTC()
|
||||
for _, o := range obs {
|
||||
country := ""
|
||||
if o.Kind == model.KindCountry {
|
||||
country = o.ID
|
||||
}
|
||||
payload, _ := json.Marshal(map[string]any{
|
||||
"id": o.ID, "kind": o.Kind, "name": o.Name,
|
||||
"metrics": o.Metrics, "note": o.Note, "src": o.Src,
|
||||
})
|
||||
s.store.Lake.Add(store.Item{
|
||||
ID: "obs:" + o.Kind + ":" + o.ID + ":" + o.Src,
|
||||
Kind: "observation",
|
||||
Source: o.Src,
|
||||
TS: now,
|
||||
Title: o.Name,
|
||||
Text: o.Note,
|
||||
Country: country,
|
||||
Payload: string(payload),
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,106 @@
|
||||
package world
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha1"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"net/url"
|
||||
"time"
|
||||
|
||||
"github.com/hanzoai/world/internal/world/store"
|
||||
)
|
||||
|
||||
// rssFetchHeaders is the single header set used for every allowlisted feed
|
||||
// fetch — by the on-demand fall-through (rss-proxy, feeds-batch) AND the
|
||||
// background warmer. One place, one behavior.
|
||||
var rssFetchHeaders = map[string]string{
|
||||
"User-Agent": browserUA,
|
||||
"Accept": "application/rss+xml, application/xml, text/xml, */*",
|
||||
}
|
||||
|
||||
// fetchFeedBody performs one bounded, allowlisted GET of a feed and returns the
|
||||
// body only when it is a usable (non-blank) 2xx. This is the ONLY live-fetch
|
||||
// path for feed bodies, shared by the request fall-through and the warmer. The
|
||||
// caller owns the timeout via ctx.
|
||||
func (s *Server) fetchFeedBody(ctx context.Context, feedURL string) ([]byte, bool) {
|
||||
body, status, err := s.getAllowlisted(ctx, feedURL, allowedRSSDomains, rssFetchHeaders)
|
||||
if err != nil || status < 200 || status >= 300 || isBlankBody(body) {
|
||||
return nil, false
|
||||
}
|
||||
return body, true
|
||||
}
|
||||
|
||||
// ingestFeedItems parses a feed body and folds its items into the searchable
|
||||
// data lake (kind=news). Cheap and write-behind (Lake.Add never blocks), so it
|
||||
// runs off both the request fall-through and the warmer without touching latency.
|
||||
func (s *Server) ingestFeedItems(feedURL string, body []byte) {
|
||||
if s.store == nil {
|
||||
return
|
||||
}
|
||||
host := feedHost(feedURL)
|
||||
for _, it := range parseFeedItems(body, feedsBatchMaxItems) {
|
||||
payload, _ := json.Marshal(map[string]any{
|
||||
"title": it.Title, "link": it.Link, "pubDate": it.PubDate,
|
||||
"source": host, "tickers": it.Tickers,
|
||||
})
|
||||
s.store.Lake.Add(store.Item{
|
||||
ID: newsItemID(it.Link, it.Title),
|
||||
Kind: "news",
|
||||
Source: host,
|
||||
TS: feedItemTime(it.PubDate),
|
||||
Title: it.Title,
|
||||
Tickers: it.Tickers,
|
||||
Payload: string(payload),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// newsItemID is the stable dedupe key for a news item: its link when present,
|
||||
// else its title. Re-ingesting the same story upserts instead of duplicating.
|
||||
func newsItemID(link, title string) string {
|
||||
seed := link
|
||||
if seed == "" {
|
||||
seed = title
|
||||
}
|
||||
sum := sha1.Sum([]byte(seed))
|
||||
return "news:" + hex.EncodeToString(sum[:])
|
||||
}
|
||||
|
||||
// feedItemTime parses a normalized RFC3339 pubDate back to a time, defaulting to
|
||||
// now when the feed omitted or mangled the date.
|
||||
func feedItemTime(pubDate string) time.Time {
|
||||
if pubDate != "" {
|
||||
if t, err := time.Parse(time.RFC3339, pubDate); err == nil {
|
||||
return t.UTC()
|
||||
}
|
||||
}
|
||||
return time.Now().UTC()
|
||||
}
|
||||
|
||||
// feedHost returns the feed's host for use as the item's source label.
|
||||
func feedHost(feedURL string) string {
|
||||
if u, err := url.Parse(feedURL); err == nil && u.Hostname() != "" {
|
||||
return u.Hostname()
|
||||
}
|
||||
return "feed"
|
||||
}
|
||||
|
||||
// curatedFeedSeed is the bootstrap warm set: the highest-value feeds behind the
|
||||
// crypto, financial-regulation, and crypto-news panels, so a brand-new pod warms
|
||||
// THEM first (before any user request). Every URL is on the RSS allowlist. After
|
||||
// the first real page load the warm set becomes demand-driven and fleet-wide; the
|
||||
// seed only bridges the very first cold boot.
|
||||
var curatedFeedSeed = []string{
|
||||
// crypto / crypto-news
|
||||
"https://www.coindesk.com/arc/outboundfeeds/rss/",
|
||||
"https://cointelegraph.com/rss",
|
||||
// financial regulation
|
||||
"https://www.sec.gov/news/pressreleases.rss",
|
||||
"https://www.federalreserve.gov/feeds/press_all.xml",
|
||||
// markets / financial
|
||||
"https://www.cnbc.com/id/100003114/device/rss/rss.html",
|
||||
"https://www.cnbc.com/id/19854910/device/rss/rss.html",
|
||||
"https://seekingalpha.com/market_currents.xml",
|
||||
"https://www.ft.com/rss/home",
|
||||
}
|
||||
@@ -0,0 +1,94 @@
|
||||
package world
|
||||
|
||||
import (
|
||||
"context"
|
||||
"math/rand"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Background feed warmer: the reason news is INSTANT.
|
||||
//
|
||||
// On boot and every interval (jittered) it fetches every warm feed, write-throughs
|
||||
// the body to the shared cache, and folds the items into the lake. So the
|
||||
// on-demand endpoints (rss-proxy, feeds-batch) ALWAYS serve from the warm cache
|
||||
// and never block a request on an upstream — stale-while-revalidate, with the
|
||||
// revalidation done here in the background instead of on the request path.
|
||||
//
|
||||
// Cross-pod dedupe is implicit: a feed whose SHARED (hanzo-kv) copy is still
|
||||
// fresh is skipped, so N pods don't all hammer the same upstream every cycle;
|
||||
// whichever pod refreshes first, the rest read its result.
|
||||
const (
|
||||
feedWarmInterval = 5 * time.Minute
|
||||
feedWarmParallel = 8
|
||||
feedWarmFetchTimeout = 10 * time.Second
|
||||
// feedWarmFreshWindow: skip refetch when a cached copy is younger than this
|
||||
// (slightly under the interval so each cycle still refreshes its own feeds).
|
||||
feedWarmFreshWindow = 4 * time.Minute
|
||||
)
|
||||
|
||||
// startFeedWarmer launches the warmer loop until ctx is cancelled. It warms once
|
||||
// shortly after boot (so a cold pod fills fast) then on the jittered interval.
|
||||
func (s *Server) startFeedWarmer(ctx context.Context) {
|
||||
go func() {
|
||||
s.warmFeeds(ctx)
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-time.After(jitter(feedWarmInterval)):
|
||||
s.warmFeeds(ctx)
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// warmFeeds refreshes every warm feed in parallel (bounded), skipping any whose
|
||||
// shared copy is still fresh. Each fetch is independently bounded; one slow
|
||||
// upstream cannot hold up the rest.
|
||||
func (s *Server) warmFeeds(ctx context.Context) {
|
||||
urls := s.feeds.WarmURLs(ctx)
|
||||
if len(urls) == 0 {
|
||||
return
|
||||
}
|
||||
sem := make(chan struct{}, feedWarmParallel)
|
||||
var wg sync.WaitGroup
|
||||
refreshed := 0
|
||||
var mu sync.Mutex
|
||||
for _, u := range urls {
|
||||
if ctx.Err() != nil {
|
||||
break
|
||||
}
|
||||
if age, ok := s.feeds.Age(ctx, u); ok && age < feedWarmFreshWindow {
|
||||
continue // a peer (or an earlier cycle) already refreshed it
|
||||
}
|
||||
wg.Add(1)
|
||||
sem <- struct{}{}
|
||||
go func(u string) {
|
||||
defer wg.Done()
|
||||
defer func() { <-sem }()
|
||||
fctx, cancel := context.WithTimeout(ctx, feedWarmFetchTimeout)
|
||||
defer cancel()
|
||||
if body, ok := s.fetchFeedBody(fctx, u); ok {
|
||||
s.feeds.Put(u, body)
|
||||
s.ingestFeedItems(u, body)
|
||||
mu.Lock()
|
||||
refreshed++
|
||||
mu.Unlock()
|
||||
}
|
||||
}(u)
|
||||
}
|
||||
wg.Wait()
|
||||
if refreshed > 0 {
|
||||
logf("world-feeds: warmed %d/%d feeds", refreshed, len(urls))
|
||||
}
|
||||
}
|
||||
|
||||
// jitter returns d spread by ±20% so pods don't synchronize their warm cycles.
|
||||
func jitter(d time.Duration) time.Duration {
|
||||
spread := int64(d) / 5 // 20%
|
||||
if spread <= 0 {
|
||||
return d
|
||||
}
|
||||
return d + time.Duration(rand.Int63n(2*spread)-spread)
|
||||
}
|
||||
@@ -0,0 +1,115 @@
|
||||
package world
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/url"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/hanzoai/world/internal/world/kv"
|
||||
)
|
||||
|
||||
const stubRSS = `<?xml version="1.0"?><rss version="2.0"><channel>
|
||||
<item><title>Bitcoin warms the cache</title><link>https://x/1</link><pubDate>Mon, 02 Jan 2006 15:04:05 -0700</pubDate></item>
|
||||
<item><title>SEC regulation update</title><link>https://x/2</link></item>
|
||||
</channel></rss>`
|
||||
|
||||
// stubFeed stands up a counting RSS upstream and allowlists its host for the
|
||||
// duration of the test (the SSRF boundary is a package var).
|
||||
func stubFeed(t *testing.T) (feedURL string, hits *int32) {
|
||||
t.Helper()
|
||||
var n int32
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
atomic.AddInt32(&n, 1)
|
||||
w.Header().Set("Content-Type", "application/xml")
|
||||
_, _ = w.Write([]byte(stubRSS))
|
||||
}))
|
||||
t.Cleanup(srv.Close)
|
||||
host := mustHost(t, srv.URL)
|
||||
allowedRSSDomains[host] = true
|
||||
t.Cleanup(func() { delete(allowedRSSDomains, host) })
|
||||
return srv.URL + "/rss.xml", &n
|
||||
}
|
||||
|
||||
// newTestServer builds a Server with the embedded store in a temp dir and
|
||||
// hanzo-kv disabled (pure per-pod cache) — hermetic, no external services.
|
||||
func newTestServer(t *testing.T) *Server {
|
||||
t.Helper()
|
||||
t.Setenv("WORLD_DATA_DIR", t.TempDir())
|
||||
t.Setenv("WORLD_KV_DISABLE", "1")
|
||||
s := NewServer()
|
||||
t.Cleanup(s.Close)
|
||||
return s
|
||||
}
|
||||
|
||||
func TestWarmerPopulatesCache(t *testing.T) {
|
||||
feedURL, hits := stubFeed(t)
|
||||
s := newTestServer(t)
|
||||
// Seed the warm set with the stub feed (demand-driven registration stand-in).
|
||||
s.feeds = NewFeedCache(kv.Open("", ""), 0, []string{feedURL})
|
||||
|
||||
s.warmFeeds(context.Background())
|
||||
|
||||
if got := atomic.LoadInt32(hits); got != 1 {
|
||||
t.Fatalf("upstream fetched %d times, want 1", got)
|
||||
}
|
||||
body, _, ok := s.feeds.Get(context.Background(), feedURL)
|
||||
if !ok || len(body) == 0 {
|
||||
t.Fatal("warmer did not populate the cache")
|
||||
}
|
||||
if string(body) != stubRSS {
|
||||
t.Fatalf("cached body mismatch")
|
||||
}
|
||||
}
|
||||
|
||||
func TestWarmerSkipsFreshFeeds(t *testing.T) {
|
||||
feedURL, hits := stubFeed(t)
|
||||
s := newTestServer(t)
|
||||
s.feeds = NewFeedCache(kv.Open("", ""), 0, nil)
|
||||
// Pre-warm: a fresh copy already in cache (age < freshWindow).
|
||||
s.feeds.Put(feedURL, []byte(stubRSS))
|
||||
|
||||
s.warmFeeds(context.Background())
|
||||
|
||||
if got := atomic.LoadInt32(hits); got != 0 {
|
||||
t.Fatalf("upstream hit %d times, want 0 (fresh feed must be skipped)", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestWarmedFeedServedInstantly is the end-to-end promise: once warmed, the
|
||||
// feeds-batch fall-through serves from cache WITHOUT any upstream fetch.
|
||||
func TestWarmedFeedServedInstantly(t *testing.T) {
|
||||
feedURL, hits := stubFeed(t)
|
||||
s := newTestServer(t)
|
||||
s.feeds = NewFeedCache(kv.Open("", ""), 0, []string{feedURL})
|
||||
s.warmFeeds(context.Background()) // one upstream fetch
|
||||
|
||||
// A subsequent request must be served from the warm cache — no new fetch.
|
||||
start := time.Now()
|
||||
body, ok, fresh := s.feedXML(context.Background(), feedURL)
|
||||
elapsed := time.Since(start)
|
||||
if !ok || fresh {
|
||||
t.Fatalf("feedXML ok=%v fresh=%v, want served-from-cache", ok, fresh)
|
||||
}
|
||||
if len(body) == 0 {
|
||||
t.Fatal("empty body from warm cache")
|
||||
}
|
||||
if got := atomic.LoadInt32(hits); got != 1 {
|
||||
t.Fatalf("upstream hit %d times, want 1 (warm read must not refetch)", got)
|
||||
}
|
||||
if elapsed > 50*time.Millisecond {
|
||||
t.Fatalf("warm read took %s, want <50ms", elapsed)
|
||||
}
|
||||
}
|
||||
|
||||
func mustHost(t *testing.T, raw string) string {
|
||||
t.Helper()
|
||||
u, err := url.Parse(raw)
|
||||
if err != nil {
|
||||
t.Fatalf("parse %q: %v", raw, err)
|
||||
}
|
||||
return u.Hostname()
|
||||
}
|
||||
@@ -0,0 +1,201 @@
|
||||
package world
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"compress/gzip"
|
||||
"context"
|
||||
"encoding/binary"
|
||||
"io"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/hanzoai/world/internal/world/kv"
|
||||
)
|
||||
|
||||
// FeedCache is the warm cache for raw RSS/Atom bodies behind the instant
|
||||
// news/feeds panels. It is a two-tier cache, ONE way to read a feed body:
|
||||
//
|
||||
// L1 — a per-pod in-memory mirror: hot reads never touch the network.
|
||||
// L2 — hanzo-kv (shared, cross-pod, survives pod restart): a warm body written
|
||||
// by any pod's warmer is instantly available to every pod, and a restarted
|
||||
// pod repopulates L1 from L2 instead of cold-starting.
|
||||
//
|
||||
// The warm-URL set (which feeds to keep fresh) is demand-driven: any feed served
|
||||
// once is registered, persisted fleet-wide in hanzo-kv, and kept fresh by the
|
||||
// background warmer. A curated seed guarantees the highest-value panels are warm
|
||||
// even on a brand-new pod before the first request. When hanzo-kv is
|
||||
// unreachable, everything degrades to the in-mem tier — still correct, just
|
||||
// per-pod and cold across restart.
|
||||
type FeedCache struct {
|
||||
mu sync.RWMutex
|
||||
mem map[string]feedRow // L1
|
||||
warm map[string]struct{} // in-mem warm-set fallback when kv is down
|
||||
kv *kv.Client
|
||||
max int
|
||||
}
|
||||
|
||||
type feedRow struct {
|
||||
body []byte
|
||||
at time.Time
|
||||
}
|
||||
|
||||
const (
|
||||
// feedKeyPrefix / warmSetKey namespace the shared hanzo-kv keys.
|
||||
feedKeyPrefix = "world:feed:v1:"
|
||||
warmSetKey = "world:feed:warm:v1"
|
||||
// feedKVTTL is the L2 safety horizon. The warmer refreshes every few minutes,
|
||||
// so a live entry is normally far younger; the TTL only bounds abandoned feeds.
|
||||
feedKVTTL = 3 * time.Hour
|
||||
// defaultFeedCacheMax bounds the L1 mirror (news feeds are small; ~500 covers
|
||||
// every frontend variant with room to spare).
|
||||
defaultFeedCacheMax = 1024
|
||||
)
|
||||
|
||||
// NewFeedCache builds the cache over an (optional) hanzo-kv client, seeding the
|
||||
// warm set with the curated bootstrap feeds so a cold pod warms them first.
|
||||
func NewFeedCache(kvc *kv.Client, max int, seed []string) *FeedCache {
|
||||
if max <= 0 {
|
||||
max = defaultFeedCacheMax
|
||||
}
|
||||
c := &FeedCache{
|
||||
mem: make(map[string]feedRow),
|
||||
warm: make(map[string]struct{}),
|
||||
kv: kvc,
|
||||
max: max,
|
||||
}
|
||||
for _, u := range seed {
|
||||
c.warm[u] = struct{}{}
|
||||
}
|
||||
// Publish the seed to the shared warm set too (best-effort), so every pod
|
||||
// converges on the same fleet-wide set.
|
||||
if len(seed) > 0 {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||||
c.kv.SAdd(ctx, warmSetKey, seed...)
|
||||
cancel()
|
||||
}
|
||||
return c
|
||||
}
|
||||
|
||||
// Get returns a cached feed body and its fetch time. L1 first (instant); on miss
|
||||
// it consults the shared L2 and, on a hit, backfills L1 so subsequent reads are
|
||||
// instant. Never touches upstream.
|
||||
func (c *FeedCache) Get(ctx context.Context, url string) ([]byte, time.Time, bool) {
|
||||
c.mu.RLock()
|
||||
row, ok := c.mem[url]
|
||||
c.mu.RUnlock()
|
||||
if ok {
|
||||
return row.body, row.at, true
|
||||
}
|
||||
if raw, ok := c.kv.GetBytes(ctx, feedKeyPrefix+url); ok {
|
||||
if at, body, ok := decodeFeed(raw); ok {
|
||||
c.storeMem(url, body, at)
|
||||
return body, at, true
|
||||
}
|
||||
}
|
||||
return nil, time.Time{}, false
|
||||
}
|
||||
|
||||
// Put write-throughs a freshly fetched body to both tiers and registers the URL
|
||||
// in the warm set (demand-driven). It uses a detached, bounded context for the
|
||||
// shared-tier writes so a write-through is never truncated by the request that
|
||||
// triggered it — durability must outlive the request. The fetch time is now.
|
||||
func (c *FeedCache) Put(url string, body []byte) {
|
||||
at := time.Now()
|
||||
c.storeMem(url, body, at)
|
||||
c.mu.Lock()
|
||||
c.warm[url] = struct{}{}
|
||||
c.mu.Unlock()
|
||||
raw := encodeFeed(at, body)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||||
defer cancel()
|
||||
c.kv.SetBytes(ctx, feedKeyPrefix+url, raw, feedKVTTL)
|
||||
c.kv.SAdd(ctx, warmSetKey, url)
|
||||
}
|
||||
|
||||
// Age returns how old the cached copy is, if any (for the warmer's
|
||||
// skip-if-fresh, cross-pod dedupe).
|
||||
func (c *FeedCache) Age(ctx context.Context, url string) (time.Duration, bool) {
|
||||
if _, at, ok := c.Get(ctx, url); ok {
|
||||
return time.Since(at), true
|
||||
}
|
||||
return 0, false
|
||||
}
|
||||
|
||||
// WarmURLs is the set of feeds to keep fresh: the fleet-wide set from hanzo-kv
|
||||
// unioned with the in-mem set (seed + demand). Deduped.
|
||||
func (c *FeedCache) WarmURLs(ctx context.Context) []string {
|
||||
set := make(map[string]struct{})
|
||||
for _, u := range c.kv.SMembers(ctx, warmSetKey) {
|
||||
set[u] = struct{}{}
|
||||
}
|
||||
c.mu.RLock()
|
||||
for u := range c.warm {
|
||||
set[u] = struct{}{}
|
||||
}
|
||||
for u := range c.mem {
|
||||
set[u] = struct{}{}
|
||||
}
|
||||
c.mu.RUnlock()
|
||||
out := make([]string, 0, len(set))
|
||||
for u := range set {
|
||||
out = append(out, u)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// Len reports the L1 mirror size (for tests/introspection).
|
||||
func (c *FeedCache) Len() int {
|
||||
c.mu.RLock()
|
||||
defer c.mu.RUnlock()
|
||||
return len(c.mem)
|
||||
}
|
||||
|
||||
// storeMem writes L1, evicting the oldest entry when at capacity.
|
||||
func (c *FeedCache) storeMem(url string, body []byte, at time.Time) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
if _, exists := c.mem[url]; !exists && len(c.mem) >= c.max {
|
||||
var oldestKey string
|
||||
var oldest time.Time
|
||||
first := true
|
||||
for k, r := range c.mem {
|
||||
if first || r.at.Before(oldest) {
|
||||
oldestKey, oldest, first = k, r.at, false
|
||||
}
|
||||
}
|
||||
if oldestKey != "" {
|
||||
delete(c.mem, oldestKey)
|
||||
}
|
||||
}
|
||||
c.mem[url] = feedRow{body: body, at: at}
|
||||
}
|
||||
|
||||
// ── L2 encoding: [8-byte unix-nano fetch time][gzip(body)] ───────────────────
|
||||
|
||||
func encodeFeed(at time.Time, body []byte) []byte {
|
||||
var buf bytes.Buffer
|
||||
var ts [8]byte
|
||||
binary.BigEndian.PutUint64(ts[:], uint64(at.UnixNano()))
|
||||
buf.Write(ts[:])
|
||||
gz := gzip.NewWriter(&buf)
|
||||
_, _ = gz.Write(body)
|
||||
_ = gz.Close()
|
||||
return buf.Bytes()
|
||||
}
|
||||
|
||||
func decodeFeed(raw []byte) (time.Time, []byte, bool) {
|
||||
if len(raw) < 8 {
|
||||
return time.Time{}, nil, false
|
||||
}
|
||||
at := time.Unix(0, int64(binary.BigEndian.Uint64(raw[:8])))
|
||||
gz, err := gzip.NewReader(bytes.NewReader(raw[8:]))
|
||||
if err != nil {
|
||||
return time.Time{}, nil, false
|
||||
}
|
||||
defer func() { _ = gz.Close() }()
|
||||
body, err := io.ReadAll(io.LimitReader(gz, maxBody))
|
||||
if err != nil {
|
||||
return time.Time{}, nil, false
|
||||
}
|
||||
return at, body, true
|
||||
}
|
||||
@@ -0,0 +1,118 @@
|
||||
package world
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/alicebob/miniredis/v2"
|
||||
"github.com/hanzoai/world/internal/world/kv"
|
||||
)
|
||||
|
||||
func TestFeedEncodeDecodeRoundTrip(t *testing.T) {
|
||||
at := time.Now().Truncate(time.Nanosecond)
|
||||
body := []byte("<rss><item>hello</item></rss>")
|
||||
gotAt, gotBody, ok := decodeFeed(encodeFeed(at, body))
|
||||
if !ok {
|
||||
t.Fatal("decode failed")
|
||||
}
|
||||
if !gotAt.Equal(at) {
|
||||
t.Fatalf("time round-trip = %v, want %v", gotAt, at)
|
||||
}
|
||||
if string(gotBody) != string(body) {
|
||||
t.Fatalf("body round-trip = %q, want %q", gotBody, body)
|
||||
}
|
||||
if _, _, ok := decodeFeed([]byte("short")); ok {
|
||||
t.Fatal("decode of truncated blob should fail")
|
||||
}
|
||||
}
|
||||
|
||||
// TestFeedCacheSharedAcrossPods proves the L2 (hanzo-kv) tier: a body written by
|
||||
// one pod is served to another pod whose L1 is cold — i.e. warming once benefits
|
||||
// the fleet, and a restarted pod (empty L1) reads a still-warm shared cache.
|
||||
func TestFeedCacheSharedAcrossPods(t *testing.T) {
|
||||
mr := miniredis.RunT(t)
|
||||
kvA := kv.Open(mr.Addr(), "")
|
||||
kvB := kv.Open(mr.Addr(), "")
|
||||
t.Cleanup(func() { kvA.Close(); kvB.Close() })
|
||||
|
||||
podA := NewFeedCache(kvA, 0, nil)
|
||||
podB := NewFeedCache(kvB, 0, nil) // separate pod: cold L1, shared L2
|
||||
|
||||
const url = "https://feeds.example.com/rss.xml"
|
||||
body := []byte("<rss><channel><item><title>Shared</title></item></channel></rss>")
|
||||
podA.Put(url, body)
|
||||
|
||||
if podB.Len() != 0 {
|
||||
t.Fatalf("podB L1 should start cold, has %d", podB.Len())
|
||||
}
|
||||
got, _, ok := podB.Get(context.Background(), url)
|
||||
if !ok || string(got) != string(body) {
|
||||
t.Fatalf("podB Get via shared L2 = %q ok=%v, want the shared body", got, ok)
|
||||
}
|
||||
if podB.Len() != 1 {
|
||||
t.Fatal("podB should have backfilled L1 from L2")
|
||||
}
|
||||
// The warm-URL set is fleet-wide (shared set), so podB knows to keep it fresh.
|
||||
if !hasURL(podB.WarmURLs(context.Background()), url) {
|
||||
t.Fatal("shared warm set missing the demand-added url")
|
||||
}
|
||||
}
|
||||
|
||||
// TestFeedCacheDegradesWithoutKV proves the graceful fallback: with hanzo-kv
|
||||
// disabled the cache is per-pod (L1 only) and correct — a "restart" (new cache)
|
||||
// is cold, never a crash.
|
||||
func TestFeedCacheDegradesWithoutKV(t *testing.T) {
|
||||
disabled := kv.Open("", "") // no hanzo-kv
|
||||
c := NewFeedCache(disabled, 0, nil)
|
||||
|
||||
const url = "https://feeds.example.com/x.xml"
|
||||
body := []byte("<rss/>")
|
||||
c.Put(url, body)
|
||||
|
||||
got, _, ok := c.Get(context.Background(), url)
|
||||
if !ok || string(got) != string(body) {
|
||||
t.Fatalf("same-pod L1 Get = %q ok=%v", got, ok)
|
||||
}
|
||||
// A fresh cache (simulated restart) with no shared L2 must miss — proving the
|
||||
// per-pod degrade is honest, not silently wrong.
|
||||
fresh := NewFeedCache(disabled, 0, nil)
|
||||
if _, _, ok := fresh.Get(context.Background(), url); ok {
|
||||
t.Fatal("restart with kv disabled should be cold, got a hit")
|
||||
}
|
||||
}
|
||||
|
||||
func TestFeedCacheSeedInWarmSet(t *testing.T) {
|
||||
c := NewFeedCache(kv.Open("", ""), 0, []string{"https://a.example/rss", "https://b.example/rss"})
|
||||
warm := c.WarmURLs(context.Background())
|
||||
if !hasURL(warm, "https://a.example/rss") || !hasURL(warm, "https://b.example/rss") {
|
||||
t.Fatalf("seed missing from warm set: %v", warm)
|
||||
}
|
||||
}
|
||||
|
||||
func TestFeedCacheEvictsOldest(t *testing.T) {
|
||||
c := NewFeedCache(kv.Open("", ""), 2, nil)
|
||||
c.Put("u1", []byte("1"))
|
||||
time.Sleep(2 * time.Millisecond)
|
||||
c.Put("u2", []byte("2"))
|
||||
time.Sleep(2 * time.Millisecond)
|
||||
c.Put("u3", []byte("3")) // over cap → evict u1 (oldest)
|
||||
if c.Len() != 2 {
|
||||
t.Fatalf("L1 size = %d, want 2 (bounded)", c.Len())
|
||||
}
|
||||
if _, _, ok := c.Get(context.Background(), "u1"); ok {
|
||||
t.Fatal("oldest entry u1 was not evicted")
|
||||
}
|
||||
if _, _, ok := c.Get(context.Background(), "u3"); !ok {
|
||||
t.Fatal("newest entry u3 missing")
|
||||
}
|
||||
}
|
||||
|
||||
func hasURL(ss []string, want string) bool {
|
||||
for _, s := range ss {
|
||||
if s == want {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -77,9 +77,12 @@ func (s *Server) handleFeedsBatch(w http.ResponseWriter, r *http.Request) {
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
defer func() { <-sem }()
|
||||
if body, ok := s.feedXML(ctx, feedURL); ok {
|
||||
if body, ok, fresh := s.feedXML(ctx, feedURL); ok {
|
||||
results[i].OK = true
|
||||
results[i].Items = parseFeedItems(body, feedsBatchMaxItems)
|
||||
if fresh {
|
||||
s.ingestFeedItems(feedURL, body) // fold a cold-miss fetch into the lake
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
@@ -91,29 +94,22 @@ func (s *Server) handleFeedsBatch(w http.ResponseWriter, r *http.Request) {
|
||||
})
|
||||
}
|
||||
|
||||
// feedXML returns the raw feed body, sharing handleRSSProxy's cache keys and
|
||||
// stale-fallback behavior.
|
||||
func (s *Server) feedXML(ctx context.Context, feedURL string) ([]byte, bool) {
|
||||
key := "rss:" + feedURL
|
||||
if v, ok := s.cache.Get(key); ok {
|
||||
return v.([]byte), true
|
||||
// feedXML returns a feed body from the shared warm cache (instant, never blocks
|
||||
// on upstream), falling through to a bounded live fetch only on a true cold miss
|
||||
// and write-through-ing the result. fresh reports whether the body came from that
|
||||
// live fetch (so the caller folds it into the lake exactly once). The warm cache
|
||||
// is the same one handleRSSProxy reads, so either path warms the other.
|
||||
func (s *Server) feedXML(ctx context.Context, feedURL string) (body []byte, ok, fresh bool) {
|
||||
if b, _, hit := s.feeds.Get(ctx, feedURL); hit {
|
||||
return b, true, false // any cached copy → serve instantly (stale-while-revalidate)
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(ctx, feedsBatchFetchTimeout)
|
||||
fctx, cancel := context.WithTimeout(ctx, feedsBatchFetchTimeout)
|
||||
defer cancel()
|
||||
body, status, err := s.getAllowlisted(ctx, feedURL, allowedRSSDomains, map[string]string{
|
||||
"User-Agent": browserUA,
|
||||
"Accept": "application/rss+xml, application/xml, text/xml, */*",
|
||||
})
|
||||
// A blank 200 is a failure, not content: never cache it (it would poison the
|
||||
// shared "rss:" key handleRSSProxy reads too). Fall back to last-good stale.
|
||||
if err != nil || status < 200 || status >= 300 || isBlankBody(body) {
|
||||
if v, ok := s.cache.GetStale(key); ok {
|
||||
return v.([]byte), true
|
||||
}
|
||||
return nil, false
|
||||
if b, okk := s.fetchFeedBody(fctx, feedURL); okk {
|
||||
s.feeds.Put(feedURL, b)
|
||||
return b, true, true
|
||||
}
|
||||
s.cache.Set(key, body, 5*time.Minute, 15*time.Minute)
|
||||
return body, true
|
||||
return nil, false, false
|
||||
}
|
||||
|
||||
// parseFeedItems handles RSS 2.0 (channel>item), RDF/RSS 1.0 (root-level
|
||||
|
||||
@@ -160,30 +160,26 @@ func (s *Server) handleRSSProxy(w http.ResponseWriter, r *http.Request) {
|
||||
writeError(w, http.StatusForbidden, "Domain not allowed")
|
||||
return
|
||||
}
|
||||
key := "rss:" + feedURL
|
||||
if v, ok := s.cache.Get(key); ok {
|
||||
writeBytes(w, http.StatusOK, "application/xml", "public, max-age=300, s-maxage=300, stale-while-revalidate=60", v.([]byte))
|
||||
const cc = "public, max-age=300, s-maxage=300, stale-while-revalidate=60"
|
||||
// Warm cache first (per-pod L1 → shared hanzo-kv L2): instant, and never blocks
|
||||
// on the upstream while ANY cached copy exists (stale-while-revalidate; the
|
||||
// background warmer does the revalidation).
|
||||
if body, _, ok := s.feeds.Get(r.Context(), feedURL); ok {
|
||||
writeBytes(w, http.StatusOK, "application/xml", cc, body)
|
||||
return
|
||||
}
|
||||
// True cold miss: bounded live fetch, write-through to the warm cache + lake.
|
||||
ctx, cancel := context.WithTimeout(r.Context(), 20*time.Second)
|
||||
defer cancel()
|
||||
body, status, err := s.getAllowlisted(ctx, feedURL, allowedRSSDomains, map[string]string{
|
||||
"User-Agent": browserUA,
|
||||
"Accept": "application/rss+xml, application/xml, text/xml, */*",
|
||||
})
|
||||
// A blank 200 is a failure, not content: never cache it (it would poison the
|
||||
// shared "rss:" key feedXML reads too). Fall back to last-good stale.
|
||||
if err != nil || status < 200 || status >= 300 || isBlankBody(body) {
|
||||
if v, ok := s.cache.GetStale(key); ok {
|
||||
writeBytes(w, http.StatusOK, "application/xml", "public, max-age=300, s-maxage=300, stale-while-revalidate=60", v.([]byte))
|
||||
return
|
||||
}
|
||||
body, ok := s.fetchFeedBody(ctx, feedURL)
|
||||
if !ok {
|
||||
w.Header().Set("Cache-Control", "no-store")
|
||||
writeJSON(w, http.StatusOK, "", map[string]any{"error": "upstream unavailable", "items": []any{}})
|
||||
return
|
||||
}
|
||||
s.cache.Set(key, body, 5*time.Minute, 15*time.Minute)
|
||||
writeBytes(w, http.StatusOK, "application/xml", "public, max-age=300, s-maxage=300, stale-while-revalidate=60", body)
|
||||
s.feeds.Put(feedURL, body)
|
||||
s.ingestFeedItems(feedURL, body)
|
||||
writeBytes(w, http.StatusOK, "application/xml", cc, body)
|
||||
}
|
||||
|
||||
// ── Hacker News ──────────────────────────────────────────────────────────────
|
||||
|
||||
@@ -0,0 +1,94 @@
|
||||
package world
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/hanzoai/world/internal/world/store"
|
||||
)
|
||||
|
||||
// The search + analytics endpoints over the ingested-data LAKE — the CTO's "one
|
||||
// place to query everything". Every ingested item (news, model observations, …)
|
||||
// is a row; search ranks them (FTS bm25 when q is present, recency otherwise),
|
||||
// analytics summarizes them. Both degrade to empty results, never 5xx.
|
||||
|
||||
// handleSearch serves GET /v1/world/search?q=&kind=&since=&country=&ticker=&limit= .
|
||||
func (s *Server) handleSearch(w http.ResponseWriter, r *http.Request) {
|
||||
if preflight(w, r, "GET, OPTIONS") || methodNotGet(w, r) {
|
||||
return
|
||||
}
|
||||
q := r.URL.Query()
|
||||
results := s.store.Lake.Search(store.SearchQuery{
|
||||
Q: q.Get("q"),
|
||||
Kind: strings.TrimSpace(q.Get("kind")),
|
||||
Country: strings.TrimSpace(q.Get("country")),
|
||||
Ticker: strings.TrimSpace(q.Get("ticker")),
|
||||
Since: parseSince(q.Get("since")),
|
||||
Limit: atoiDefault(q.Get("limit"), 30),
|
||||
})
|
||||
out := make([]map[string]any, 0, len(results))
|
||||
for _, it := range results {
|
||||
m := map[string]any{
|
||||
"id": it.ID, "kind": it.Kind, "source": it.Source,
|
||||
"ts": it.TS.Format(time.RFC3339), "title": it.Title,
|
||||
}
|
||||
if it.Text != "" {
|
||||
m["text"] = it.Text
|
||||
}
|
||||
if len(it.Tickers) > 0 {
|
||||
m["tickers"] = it.Tickers
|
||||
}
|
||||
if it.Country != "" {
|
||||
m["country"] = it.Country
|
||||
}
|
||||
if it.HasGeo {
|
||||
m["lat"], m["lon"] = it.Lat, it.Lon
|
||||
}
|
||||
if it.Payload != "" {
|
||||
m["payload"] = json.RawMessage(it.Payload)
|
||||
}
|
||||
out = append(out, m)
|
||||
}
|
||||
writeJSON(w, http.StatusOK, "public, max-age=15, s-maxage=15, stale-while-revalidate=60", map[string]any{
|
||||
"query": q.Get("q"), "count": len(out), "results": out, "updatedAt": nowRFC(),
|
||||
})
|
||||
}
|
||||
|
||||
// handleAnalytics serves GET /v1/world/analytics?hours= — a cross-cutting
|
||||
// summary of the lake: totals and breakdowns by kind, source, and top tickers.
|
||||
func (s *Server) handleAnalytics(w http.ResponseWriter, r *http.Request) {
|
||||
if preflight(w, r, "GET, OPTIONS") || methodNotGet(w, r) {
|
||||
return
|
||||
}
|
||||
hours := clampInt(r.URL.Query().Get("hours"), 24, 1, 720)
|
||||
writeJSON(w, http.StatusOK, "public, max-age=30, s-maxage=30, stale-while-revalidate=120",
|
||||
s.store.Lake.Analytics(hours))
|
||||
}
|
||||
|
||||
// parseSince accepts an absolute RFC3339 timestamp, a "<n>d" day window, a Go
|
||||
// duration ("24h","90m"), or a bare integer of hours. Unparseable/empty → no
|
||||
// lower bound.
|
||||
func parseSince(raw string) time.Time {
|
||||
raw = strings.TrimSpace(raw)
|
||||
if raw == "" {
|
||||
return time.Time{}
|
||||
}
|
||||
if t, err := time.Parse(time.RFC3339, raw); err == nil {
|
||||
return t
|
||||
}
|
||||
if strings.HasSuffix(raw, "d") {
|
||||
if n, err := strconv.Atoi(strings.TrimSuffix(raw, "d")); err == nil && n > 0 {
|
||||
return time.Now().Add(-time.Duration(n) * 24 * time.Hour)
|
||||
}
|
||||
}
|
||||
if d, err := time.ParseDuration(raw); err == nil && d > 0 {
|
||||
return time.Now().Add(-d)
|
||||
}
|
||||
if n, err := strconv.Atoi(raw); err == nil && n > 0 {
|
||||
return time.Now().Add(-time.Duration(n) * time.Hour)
|
||||
}
|
||||
return time.Time{}
|
||||
}
|
||||
@@ -0,0 +1,95 @@
|
||||
package world
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// serve drives one handler with an in-memory request/response and returns the
|
||||
// decoded JSON body plus status.
|
||||
func serve(t *testing.T, h http.HandlerFunc, method, target, bearer, body string) (map[string]any, int) {
|
||||
t.Helper()
|
||||
var r *http.Request
|
||||
if body != "" {
|
||||
r = httptest.NewRequest(method, target, strings.NewReader(body))
|
||||
} else {
|
||||
r = httptest.NewRequest(method, target, nil)
|
||||
}
|
||||
if bearer != "" {
|
||||
r.Header.Set("Authorization", "Bearer "+bearer)
|
||||
}
|
||||
rec := httptest.NewRecorder()
|
||||
h(rec, r)
|
||||
res := rec.Result()
|
||||
defer res.Body.Close()
|
||||
raw, _ := io.ReadAll(res.Body)
|
||||
var m map[string]any
|
||||
if len(raw) > 0 {
|
||||
_ = json.Unmarshal(raw, &m)
|
||||
}
|
||||
return m, res.StatusCode
|
||||
}
|
||||
|
||||
func TestSearchEndpointEmptyNever5xx(t *testing.T) {
|
||||
s := newTestServer(t)
|
||||
m, code := serve(t, s.handleSearch, http.MethodGet, "/v1/world/search", "", "")
|
||||
if code != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200", code)
|
||||
}
|
||||
if m["count"].(float64) != 0 {
|
||||
t.Fatalf("empty lake count = %v, want 0", m["count"])
|
||||
}
|
||||
if _, ok := m["results"].([]any); !ok {
|
||||
t.Fatalf("results not an array: %v", m["results"])
|
||||
}
|
||||
}
|
||||
|
||||
func TestSearchAndAnalyticsReturnIngested(t *testing.T) {
|
||||
s := newTestServer(t)
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
go s.store.Lake.Run(ctx) // write-behind consumer
|
||||
|
||||
// Ingest via the same path the feed handlers use.
|
||||
s.ingestFeedItems("https://cointelegraph.com/rss", []byte(stubRSS))
|
||||
|
||||
// Poll until the write-behind flush lands (≤1s tick).
|
||||
var m map[string]any
|
||||
deadline := time.Now().Add(4 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
var code int
|
||||
m, code = serve(t, s.handleSearch, http.MethodGet, "/v1/world/search?q=bitcoin", "", "")
|
||||
if code != http.StatusOK {
|
||||
t.Fatalf("search status = %d", code)
|
||||
}
|
||||
if m["count"].(float64) >= 1 {
|
||||
break
|
||||
}
|
||||
time.Sleep(40 * time.Millisecond)
|
||||
}
|
||||
if m["count"].(float64) < 1 {
|
||||
t.Fatalf("ingested news not searchable: %v", m)
|
||||
}
|
||||
first := m["results"].([]any)[0].(map[string]any)
|
||||
if !strings.Contains(strings.ToLower(first["title"].(string)), "bitcoin") {
|
||||
t.Fatalf("top result title = %v, want a Bitcoin item", first["title"])
|
||||
}
|
||||
if first["kind"] != "news" {
|
||||
t.Fatalf("kind = %v, want news", first["kind"])
|
||||
}
|
||||
|
||||
// Analytics reflects the same ingest.
|
||||
a, code := serve(t, s.handleAnalytics, http.MethodGet, "/v1/world/analytics", "", "")
|
||||
if code != http.StatusOK {
|
||||
t.Fatalf("analytics status = %d", code)
|
||||
}
|
||||
if a["total"].(float64) < 1 {
|
||||
t.Fatalf("analytics total = %v, want ≥1", a["total"])
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,110 @@
|
||||
package world
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"github.com/hanzoai/world/internal/world/store"
|
||||
)
|
||||
|
||||
// Per-identity settings: server-side dashboard sync for signed-in users.
|
||||
//
|
||||
// The frontend layout engine persists panel geometry / layout mode / preferences
|
||||
// to localStorage today. These endpoints let a follow-up sync a SIGNED-IN user's
|
||||
// dashboard across devices: the blob is keyed by (org, user_sub, project) from
|
||||
// the IAM bearer, so it is isolated per identity. Anonymous callers get 401 and
|
||||
// keep using localStorage — no server state for them.
|
||||
//
|
||||
// FRONTEND HOOK (another agent owns src/): after the layout store mutates, if the
|
||||
// user is signed in, PUT the same JSON it writes to localStorage to
|
||||
// /v1/world/settings (Authorization: Bearer <token>); on load, if signed in, GET
|
||||
// /v1/world/settings and prefer a non-empty server blob over localStorage. One
|
||||
// debounced PUT on change, one GET on boot — the endpoints below are the whole API.
|
||||
|
||||
// wIdentity is the caller's IAM identity resolved from userinfo.
|
||||
type wIdentity struct {
|
||||
Org string `json:"owner"`
|
||||
Sub string `json:"sub"`
|
||||
}
|
||||
|
||||
// introspectIdentity resolves the caller's org (owner claim) + subject from IAM
|
||||
// userinfo, memoized by token hash for a short TTL (mirrors introspectOwner). It
|
||||
// is the authoritative, IAM-signed identity — never a client-supplied header.
|
||||
func (s *Server) introspectIdentity(ctx context.Context, bearer string) (wIdentity, error) {
|
||||
sum := sha256.Sum256([]byte(bearer))
|
||||
key := "identity:" + hex.EncodeToString(sum[:12])
|
||||
if v, ok := s.cache.Get(key); ok {
|
||||
return v.(wIdentity), nil
|
||||
}
|
||||
var id wIdentity
|
||||
if err := s.getJSON(ctx, iamIssuer()+"/v1/iam/oauth/userinfo",
|
||||
map[string]string{"Authorization": bearer}, &id); err != nil {
|
||||
return wIdentity{}, err
|
||||
}
|
||||
s.cache.Set(key, id, 60*time.Second, 60*time.Second)
|
||||
return id, nil
|
||||
}
|
||||
|
||||
// handleSettings serves GET (read this identity's blob) and PUT (upsert it),
|
||||
// both bearer-gated. Never 5xx: a degraded store returns {} on GET and ok:false
|
||||
// on PUT so the client falls back to localStorage.
|
||||
func (s *Server) handleSettings(w http.ResponseWriter, r *http.Request) {
|
||||
setCORS(w, "GET, PUT, OPTIONS")
|
||||
if r.Method == http.MethodOptions {
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
return
|
||||
}
|
||||
if r.Method != http.MethodGet && r.Method != http.MethodPut {
|
||||
writeError(w, http.StatusMethodNotAllowed, "GET or PUT")
|
||||
return
|
||||
}
|
||||
bearer := userBearer(r)
|
||||
if bearer == "" {
|
||||
writeError(w, http.StatusUnauthorized, "Sign in required")
|
||||
return
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
|
||||
defer cancel()
|
||||
id, err := s.introspectIdentity(ctx, bearer)
|
||||
if err != nil || id.Sub == "" {
|
||||
writeError(w, http.StatusUnauthorized, "Sign in required")
|
||||
return
|
||||
}
|
||||
ident := store.Identity{Org: id.Org, UserSub: id.Sub, Project: r.URL.Query().Get("project")}
|
||||
|
||||
if r.Method == http.MethodGet {
|
||||
blob, ok := s.store.Settings.Get(ident)
|
||||
if !ok {
|
||||
blob = json.RawMessage(`{}`)
|
||||
}
|
||||
writeJSON(w, http.StatusOK, "private, no-store", map[string]any{"settings": blob})
|
||||
return
|
||||
}
|
||||
|
||||
// PUT: validate the body is a JSON object at the boundary, then upsert.
|
||||
raw, err := io.ReadAll(http.MaxBytesReader(w, r.Body, 256<<10))
|
||||
if err != nil {
|
||||
writeError(w, http.StatusBadRequest, "Body too large")
|
||||
return
|
||||
}
|
||||
if !json.Valid(raw) || !isJSONObject(raw) {
|
||||
writeError(w, http.StatusBadRequest, "Body must be a JSON object")
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, "private, no-store", map[string]any{
|
||||
"ok": s.store.Settings.Put(ident, json.RawMessage(raw)),
|
||||
})
|
||||
}
|
||||
|
||||
// isJSONObject reports whether raw is a JSON object ({...}), not an array or
|
||||
// scalar — settings are always an object blob.
|
||||
func isJSONObject(raw []byte) bool {
|
||||
raw = bytes.TrimSpace(raw)
|
||||
return len(raw) > 0 && raw[0] == '{'
|
||||
}
|
||||
@@ -0,0 +1,115 @@
|
||||
package world
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// stubIAM stands up a userinfo endpoint that maps a bearer token to an IAM
|
||||
// identity, and points the server's issuer at it. This exercises the real
|
||||
// introspectIdentity path hermetically (no live IAM).
|
||||
func stubIAM(t *testing.T, tokenToIdentity map[string]wIdentity) {
|
||||
t.Helper()
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.URL.Path != "/v1/iam/oauth/userinfo" {
|
||||
http.NotFound(w, r)
|
||||
return
|
||||
}
|
||||
tok := r.Header.Get("Authorization")
|
||||
id, ok := tokenToIdentity[tok]
|
||||
if !ok {
|
||||
w.WriteHeader(http.StatusUnauthorized)
|
||||
return
|
||||
}
|
||||
_ = json.NewEncoder(w).Encode(id)
|
||||
}))
|
||||
t.Cleanup(srv.Close)
|
||||
t.Setenv("HANZO_IAM_ISSUER", srv.URL)
|
||||
}
|
||||
|
||||
func TestSettingsAnonymousUnauthorized(t *testing.T) {
|
||||
s := newTestServer(t)
|
||||
_, code := serve(t, s.handleSettings, http.MethodGet, "/v1/world/settings", "", "")
|
||||
if code != http.StatusUnauthorized {
|
||||
t.Fatalf("anonymous GET status = %d, want 401", code)
|
||||
}
|
||||
_, code = serve(t, s.handleSettings, http.MethodPut, "/v1/world/settings", "", `{"a":1}`)
|
||||
if code != http.StatusUnauthorized {
|
||||
t.Fatalf("anonymous PUT status = %d, want 401", code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSettingsUpsertGetPerIdentity(t *testing.T) {
|
||||
s := newTestServer(t)
|
||||
stubIAM(t, map[string]wIdentity{
|
||||
"Bearer alice": {Org: "acme", Sub: "alice"},
|
||||
"Bearer bob": {Org: "acme", Sub: "bob"},
|
||||
})
|
||||
|
||||
// Alice stores her dashboard.
|
||||
m, code := serve(t, s.handleSettings, http.MethodPut, "/v1/world/settings", "alice", `{"layout":"grid","cell":40}`)
|
||||
if code != http.StatusOK || m["ok"] != true {
|
||||
t.Fatalf("alice PUT = %d %v", code, m)
|
||||
}
|
||||
// Bob stores a different one.
|
||||
if m, code := serve(t, s.handleSettings, http.MethodPut, "/v1/world/settings", "bob", `{"layout":"list"}`); code != http.StatusOK || m["ok"] != true {
|
||||
t.Fatalf("bob PUT = %d %v", code, m)
|
||||
}
|
||||
|
||||
// Each reads back exactly their own — server-side, cross-device, isolated.
|
||||
m, code = serve(t, s.handleSettings, http.MethodGet, "/v1/world/settings", "alice", "")
|
||||
if code != http.StatusOK {
|
||||
t.Fatalf("alice GET status = %d", code)
|
||||
}
|
||||
if got, want := jsonStr(t, m["settings"]), normJSON(t, `{"layout":"grid","cell":40}`); got != want {
|
||||
t.Fatalf("alice settings = %s, want %s", got, want)
|
||||
}
|
||||
m, _ = serve(t, s.handleSettings, http.MethodGet, "/v1/world/settings", "bob", "")
|
||||
if got, want := jsonStr(t, m["settings"]), normJSON(t, `{"layout":"list"}`); got != want {
|
||||
t.Fatalf("bob settings = %s, want %s (identity isolation broken)", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSettingsRejectsNonObject(t *testing.T) {
|
||||
s := newTestServer(t)
|
||||
stubIAM(t, map[string]wIdentity{"Bearer alice": {Org: "acme", Sub: "alice"}})
|
||||
_, code := serve(t, s.handleSettings, http.MethodPut, "/v1/world/settings", "alice", `[1,2,3]`)
|
||||
if code != http.StatusBadRequest {
|
||||
t.Fatalf("array body status = %d, want 400", code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSettingsMissingIsEmptyObject(t *testing.T) {
|
||||
s := newTestServer(t)
|
||||
stubIAM(t, map[string]wIdentity{"Bearer newuser": {Org: "acme", Sub: "newuser"}})
|
||||
m, code := serve(t, s.handleSettings, http.MethodGet, "/v1/world/settings", "newuser", "")
|
||||
if code != http.StatusOK {
|
||||
t.Fatalf("status = %d", code)
|
||||
}
|
||||
if got := jsonStr(t, m["settings"]); got != `{}` {
|
||||
t.Fatalf("absent settings = %s, want {}", got)
|
||||
}
|
||||
}
|
||||
|
||||
func jsonStr(t *testing.T, v any) string {
|
||||
t.Helper()
|
||||
b, err := json.Marshal(v)
|
||||
if err != nil {
|
||||
t.Fatalf("marshal: %v", err)
|
||||
}
|
||||
return string(b)
|
||||
}
|
||||
|
||||
// normJSON round-trips a JSON literal through decode+encode so map key ordering
|
||||
// matches jsonStr's output (Go marshals map keys sorted) — order-independent
|
||||
// comparison.
|
||||
func normJSON(t *testing.T, s string) string {
|
||||
t.Helper()
|
||||
var v any
|
||||
if err := json.Unmarshal([]byte(s), &v); err != nil {
|
||||
t.Fatalf("unmarshal %q: %v", s, err)
|
||||
}
|
||||
return jsonStr(t, v)
|
||||
}
|
||||
@@ -0,0 +1,136 @@
|
||||
// Package kv is world's thin, graceful-degrade client for hanzo-kv — the shared
|
||||
// Valkey/Redis hot cache (k8s Service hanzo-kv:6379, ns hanzo). It exists so the
|
||||
// feed/response warm cache is SHARED across all world pods and survives a pod
|
||||
// restart: warming once benefits the whole fleet, and a restarted pod reads a
|
||||
// still-warm cache instead of cold-starting.
|
||||
//
|
||||
// Every method degrades cleanly — a nil/unconfigured client, an unreachable
|
||||
// server, or any transport error yields a clean miss / no-op, never a blocking
|
||||
// call or an error the caller must handle. A tiny circuit breaker parks a
|
||||
// downed server for a cooldown so the hot path is not repeatedly stalled dialing
|
||||
// a dead host. The queryable data lake lives in SQLite (package store); this is
|
||||
// only the speed layer in front of feeds.
|
||||
package kv
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
// breakerCooldown parks a failing server: after an error, ops short-circuit to
|
||||
// "miss" for this long instead of re-dialing on every request.
|
||||
const breakerCooldown = 15 * time.Second
|
||||
|
||||
// Client wraps a go-redis client. A zero/disabled Client (r == nil) is valid and
|
||||
// behaves as a permanent clean miss, so local dev and CI need no Redis.
|
||||
type Client struct {
|
||||
r *redis.Client
|
||||
downUntil atomic.Int64 // unix-nano; server parked until then
|
||||
}
|
||||
|
||||
// Open builds a client for addr (e.g. "hanzo-kv:6379"). An empty addr returns a
|
||||
// disabled client (pure miss) — the correct behavior for environments without
|
||||
// hanzo-kv. Timeouts are short so a slow/dead server degrades fast; retries are
|
||||
// disabled because we fail over to the embedded/in-mem cache, not by retrying.
|
||||
func Open(addr, password string) *Client {
|
||||
if addr == "" {
|
||||
return &Client{}
|
||||
}
|
||||
return &Client{r: redis.NewClient(&redis.Options{
|
||||
Addr: addr,
|
||||
Password: password,
|
||||
DialTimeout: 2 * time.Second,
|
||||
ReadTimeout: time.Second,
|
||||
WriteTimeout: time.Second,
|
||||
PoolSize: 8,
|
||||
MaxRetries: -1, // fail fast; degrade rather than retry
|
||||
})}
|
||||
}
|
||||
|
||||
// Enabled reports whether a server is configured (not whether it is currently
|
||||
// reachable).
|
||||
func (c *Client) Enabled() bool { return c != nil && c.r != nil }
|
||||
|
||||
func (c *Client) available() bool {
|
||||
if c == nil || c.r == nil {
|
||||
return false
|
||||
}
|
||||
return time.Now().UnixNano() >= c.downUntil.Load()
|
||||
}
|
||||
|
||||
// trip parks the server for the cooldown after a transport failure.
|
||||
func (c *Client) trip() { c.downUntil.Store(time.Now().Add(breakerCooldown).UnixNano()) }
|
||||
|
||||
// GetBytes returns the value for key, or (nil,false) on miss/failure. A real
|
||||
// cache miss (redis.Nil) does not trip the breaker; a transport error does.
|
||||
func (c *Client) GetBytes(ctx context.Context, key string) ([]byte, bool) {
|
||||
if !c.available() {
|
||||
return nil, false
|
||||
}
|
||||
b, err := c.r.Get(ctx, key).Bytes()
|
||||
if err == redis.Nil {
|
||||
return nil, false
|
||||
}
|
||||
if err != nil {
|
||||
c.trip()
|
||||
return nil, false
|
||||
}
|
||||
return b, true
|
||||
}
|
||||
|
||||
// SetBytes writes key=val with a TTL. Best-effort: failures trip the breaker and
|
||||
// are otherwise ignored (the value is still cached in the per-pod mirror).
|
||||
func (c *Client) SetBytes(ctx context.Context, key string, val []byte, ttl time.Duration) {
|
||||
if !c.available() {
|
||||
return
|
||||
}
|
||||
if err := c.r.Set(ctx, key, val, ttl).Err(); err != nil {
|
||||
c.trip()
|
||||
}
|
||||
}
|
||||
|
||||
// SAdd adds members to a set (the fleet-wide warm-URL registry). Best-effort.
|
||||
func (c *Client) SAdd(ctx context.Context, key string, members ...string) {
|
||||
if !c.available() || len(members) == 0 {
|
||||
return
|
||||
}
|
||||
vals := make([]any, len(members))
|
||||
for i, m := range members {
|
||||
vals[i] = m
|
||||
}
|
||||
if err := c.r.SAdd(ctx, key, vals...).Err(); err != nil {
|
||||
c.trip()
|
||||
}
|
||||
}
|
||||
|
||||
// SMembers returns the set members, or nil on miss/failure.
|
||||
func (c *Client) SMembers(ctx context.Context, key string) []string {
|
||||
if !c.available() {
|
||||
return nil
|
||||
}
|
||||
v, err := c.r.SMembers(ctx, key).Result()
|
||||
if err != nil {
|
||||
c.trip()
|
||||
return nil
|
||||
}
|
||||
return v
|
||||
}
|
||||
|
||||
// Ping checks reachability (used once at boot to log status). Returns an error
|
||||
// when disabled or unreachable.
|
||||
func (c *Client) Ping(ctx context.Context) error {
|
||||
if c == nil || c.r == nil {
|
||||
return redis.ErrClosed
|
||||
}
|
||||
return c.r.Ping(ctx).Err()
|
||||
}
|
||||
|
||||
// Close releases the connection pool. Safe on a disabled client.
|
||||
func (c *Client) Close() {
|
||||
if c != nil && c.r != nil {
|
||||
_ = c.r.Close()
|
||||
}
|
||||
}
|
||||
@@ -25,9 +25,19 @@ type Engine struct {
|
||||
snapPath string
|
||||
history *History
|
||||
|
||||
// sink, when set, receives every cycle's raw observations after they fold.
|
||||
// It lets an owner (package world) dump observations into the queryable data
|
||||
// lake WITHOUT this package knowing anything about storage — the engine stays
|
||||
// decomplected from the datastore; it just calls a value-in hook.
|
||||
sink func([]Observation)
|
||||
|
||||
startOnce sync.Once
|
||||
}
|
||||
|
||||
// SetObservationSink registers a hook called with each cycle's observations after
|
||||
// they are folded. Set once before Start; nil (the default) is a no-op.
|
||||
func (e *Engine) SetObservationSink(fn func([]Observation)) { e.sink = fn }
|
||||
|
||||
// New builds an engine. dataDir is where the warm-start snapshot and the history
|
||||
// ring live; interval<=0 uses DefaultInterval.
|
||||
func New(sources []Source, dataDir string, interval time.Duration) *Engine {
|
||||
@@ -114,6 +124,9 @@ func (e *Engine) IngestOnce(ctx context.Context) {
|
||||
wg.Wait()
|
||||
changes := e.store.Apply(all, time.Now().UTC())
|
||||
log.Printf("world-model: ingest folded %d observations, %d changes", len(all), len(changes))
|
||||
if e.sink != nil && len(all) > 0 {
|
||||
e.sink(all)
|
||||
}
|
||||
}
|
||||
|
||||
// ── snapshot ─────────────────────────────────────────────────────────────────
|
||||
|
||||
@@ -65,6 +65,15 @@ func (s *Server) mount(mux registrar) {
|
||||
mux.HandleFunc("/v1/world/youtube/live", s.handleYouTubeLive)
|
||||
mux.HandleFunc("/v1/world/youtube/embed", s.handleYouTubeEmbed)
|
||||
|
||||
// ingested-data lake — the "one place to query everything" (search +
|
||||
// analytics across ALL ingested items: news, model observations, …).
|
||||
mux.HandleFunc("/v1/world/search", s.handleSearch)
|
||||
mux.HandleFunc("/v1/world/analytics", s.handleAnalytics)
|
||||
|
||||
// per-identity settings — server-side dashboard sync for signed-in users
|
||||
// (bearer-gated; anonymous keeps localStorage).
|
||||
mux.HandleFunc("/v1/world/settings", s.handleSettings)
|
||||
|
||||
// econ / humanitarian
|
||||
mux.HandleFunc("/v1/world/fred-data", s.handleFRED)
|
||||
mux.HandleFunc("/v1/world/worldbank", s.handleWorldBank)
|
||||
|
||||
@@ -21,8 +21,10 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/hanzoai/world/internal/world/kv"
|
||||
"github.com/hanzoai/world/internal/world/mcp"
|
||||
"github.com/hanzoai/world/internal/world/model"
|
||||
"github.com/hanzoai/world/internal/world/store"
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -42,10 +44,18 @@ type Server struct {
|
||||
ai *AIClient
|
||||
worldModel *model.Engine
|
||||
mcp *mcp.Server
|
||||
|
||||
// Datastore layer (see datastore.go): kv is the shared hanzo-kv hot cache,
|
||||
// feeds is the two-tier warm feed-body cache in front of it, and store is the
|
||||
// embedded SQLite lake + per-identity settings.
|
||||
kv *kv.Client
|
||||
store *store.DB
|
||||
feeds *FeedCache
|
||||
}
|
||||
|
||||
// NewServer constructs the backend and its world-model engine (built from the
|
||||
// feed sources in model_sources.go). Call StartModel to begin ingest.
|
||||
// NewServer constructs the backend, its world-model engine (built from the feed
|
||||
// sources in model_sources.go), and the datastore layer. Call StartModel and
|
||||
// StartDatastore to begin the background loops.
|
||||
func NewServer() *Server {
|
||||
s := &Server{
|
||||
client: &http.Client{Timeout: 25 * time.Second},
|
||||
@@ -54,6 +64,7 @@ func NewServer() *Server {
|
||||
mcp: mcp.New(),
|
||||
}
|
||||
s.worldModel = model.New(s.modelSources(), modelDataDir(), modelInterval())
|
||||
s.initDatastore()
|
||||
return s
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,394 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Item is one normalized, queryable record in the ingested-data lake. Every
|
||||
// upstream — news/feed items, events, indicators, market snapshots, model
|
||||
// observations — folds into this ONE shape so everything is searchable and
|
||||
// countable together. ID is the stable dedupe key: re-ingesting the same item
|
||||
// upserts (no duplicates), so the warmer re-touching a feed every few minutes
|
||||
// keeps the lake fresh without growing it.
|
||||
type Item struct {
|
||||
ID string `json:"id"`
|
||||
Kind string `json:"kind"` // news | event | indicator | market | observation
|
||||
Source string `json:"source"` // feed host, provider, or model source
|
||||
TS time.Time `json:"ts"` // the item's own timestamp
|
||||
Title string `json:"title"`
|
||||
Text string `json:"text,omitempty"`
|
||||
Tickers []string `json:"tickers,omitempty"`
|
||||
Country string `json:"country,omitempty"` // ISO code or ""
|
||||
Lat float64 `json:"lat,omitempty"`
|
||||
Lon float64 `json:"lon,omitempty"`
|
||||
HasGeo bool `json:"-"`
|
||||
Payload string `json:"payload,omitempty"` // compact original JSON
|
||||
}
|
||||
|
||||
// Lake is the write-behind ingest sink and the query surface (search +
|
||||
// analytics) over the items table. Writes are buffered and flushed in batched
|
||||
// transactions by Run so the request path never blocks on disk; reads go
|
||||
// straight to SQLite (single serialized connection, sub-ms).
|
||||
type Lake struct {
|
||||
db *sql.DB // nil in degraded mode
|
||||
ch chan Item
|
||||
retention time.Duration
|
||||
}
|
||||
|
||||
// lakeBuffer bounds the write-behind queue. Full → drop (the lake is a
|
||||
// derived cache of the feeds/model, never the source of truth).
|
||||
const lakeBuffer = 8192
|
||||
|
||||
func newLake(db *sql.DB, retention time.Duration) *Lake {
|
||||
return &Lake{db: db, ch: make(chan Item, lakeBuffer), retention: retention}
|
||||
}
|
||||
|
||||
// Add enqueues an item for write-behind persistence. Non-blocking: if the
|
||||
// buffer is full (writer stalled) the item is dropped rather than slowing the
|
||||
// caller. No-op in degraded mode or without an ID.
|
||||
func (l *Lake) Add(it Item) {
|
||||
if l == nil || l.db == nil || it.ID == "" {
|
||||
return
|
||||
}
|
||||
select {
|
||||
case l.ch <- it:
|
||||
default: // buffer full — drop; the source will re-emit on its next cycle
|
||||
}
|
||||
}
|
||||
|
||||
// Run is the write-behind consumer + retention prune loop. It batches queued
|
||||
// items into periodic transactions and prunes expired rows hourly, until ctx is
|
||||
// cancelled (final flush on the way out). Start once from the server lifecycle.
|
||||
func (l *Lake) Run(ctx context.Context) {
|
||||
if l == nil || l.db == nil {
|
||||
return
|
||||
}
|
||||
batch := make([]Item, 0, 256)
|
||||
flush := func() {
|
||||
if len(batch) == 0 {
|
||||
return
|
||||
}
|
||||
if err := l.insert(batch); err != nil {
|
||||
logStore("lake insert (%d items): %v", len(batch), err)
|
||||
}
|
||||
batch = batch[:0]
|
||||
}
|
||||
tick := time.NewTicker(time.Second)
|
||||
defer tick.Stop()
|
||||
prune := time.NewTicker(time.Hour)
|
||||
defer prune.Stop()
|
||||
// Prune once shortly after boot so a restart trims yesterday's backlog.
|
||||
l.Prune(l.retention)
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
flush()
|
||||
return
|
||||
case it := <-l.ch:
|
||||
batch = append(batch, it)
|
||||
if len(batch) >= 256 {
|
||||
flush()
|
||||
}
|
||||
case <-tick.C:
|
||||
flush()
|
||||
case <-prune.C:
|
||||
l.Prune(l.retention)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// insert upserts a batch in one transaction. On conflict it refreshes the
|
||||
// mutable fields but preserves created_at (first-seen) so retention measures
|
||||
// true age. The FTS index follows via triggers.
|
||||
func (l *Lake) insert(batch []Item) error {
|
||||
tx, err := l.db.Begin()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
stmt, err := tx.Prepare(`INSERT INTO items
|
||||
(id, kind, source, ts, title, text, tickers, country, lat, lon, payload, created_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
ON CONFLICT(id) DO UPDATE SET
|
||||
kind=excluded.kind, source=excluded.source, ts=excluded.ts,
|
||||
title=excluded.title, text=excluded.text, tickers=excluded.tickers,
|
||||
country=excluded.country, lat=excluded.lat, lon=excluded.lon,
|
||||
payload=excluded.payload`)
|
||||
if err != nil {
|
||||
_ = tx.Rollback()
|
||||
return err
|
||||
}
|
||||
defer func() { _ = stmt.Close() }()
|
||||
now := time.Now().Unix()
|
||||
for _, it := range batch {
|
||||
var lat, lon any
|
||||
if it.HasGeo {
|
||||
lat, lon = it.Lat, it.Lon
|
||||
}
|
||||
if _, err := stmt.Exec(it.ID, it.Kind, it.Source, it.TS.Unix(), it.Title, it.Text,
|
||||
packTickers(it.Tickers), it.Country, lat, lon, it.Payload, now); err != nil {
|
||||
_ = tx.Rollback()
|
||||
return err
|
||||
}
|
||||
}
|
||||
return tx.Commit()
|
||||
}
|
||||
|
||||
// SearchQuery is a query across the whole lake. Empty Q browses by recency;
|
||||
// non-empty Q ranks by FTS bm25. All filters are optional and compose.
|
||||
type SearchQuery struct {
|
||||
Q string
|
||||
Kind string
|
||||
Country string
|
||||
Ticker string
|
||||
Since time.Time
|
||||
Limit int
|
||||
}
|
||||
|
||||
// Search returns matching items, ranked by relevance (with Q) or recency
|
||||
// (without). Never errors out: any failure yields an empty slice (never-5xx).
|
||||
func (l *Lake) Search(q SearchQuery) []Item {
|
||||
if l == nil || l.db == nil {
|
||||
return []Item{}
|
||||
}
|
||||
limit := q.Limit
|
||||
if limit <= 0 || limit > 100 {
|
||||
limit = 30
|
||||
}
|
||||
// The items table is ALWAYS aliased `i` (bare browse or FTS join), so every
|
||||
// filter and the projection reference i.<col> uniformly — no per-branch column
|
||||
// rewriting. Free text goes through the FTS index (bm25 ranked); its absence
|
||||
// browses by recency. Both share the same filter + projection code below.
|
||||
const cols = "i.id,i.kind,i.source,i.ts,i.title,i.text,i.tickers,i.country,i.lat,i.lon,i.payload"
|
||||
var (
|
||||
sb strings.Builder
|
||||
args []any
|
||||
)
|
||||
fts := ftsQuery(q.Q)
|
||||
if fts != "" {
|
||||
sb.WriteString("SELECT " + cols + " FROM items_fts f JOIN items i ON i.rowid=f.rowid WHERE items_fts MATCH ?")
|
||||
args = append(args, fts)
|
||||
} else {
|
||||
sb.WriteString("SELECT " + cols + " FROM items i WHERE 1=1")
|
||||
}
|
||||
if q.Kind != "" {
|
||||
sb.WriteString(" AND i.kind=?")
|
||||
args = append(args, q.Kind)
|
||||
}
|
||||
if q.Country != "" {
|
||||
sb.WriteString(" AND i.country=?")
|
||||
args = append(args, strings.ToUpper(q.Country))
|
||||
}
|
||||
if !q.Since.IsZero() {
|
||||
sb.WriteString(" AND i.ts>=?")
|
||||
args = append(args, q.Since.Unix())
|
||||
}
|
||||
if t := strings.ToLower(strings.TrimSpace(q.Ticker)); t != "" {
|
||||
sb.WriteString(" AND i.tickers LIKE ?")
|
||||
args = append(args, "% "+t+" %")
|
||||
}
|
||||
if fts != "" {
|
||||
sb.WriteString(" ORDER BY bm25(items_fts) LIMIT ?")
|
||||
} else {
|
||||
sb.WriteString(" ORDER BY i.ts DESC LIMIT ?")
|
||||
}
|
||||
args = append(args, limit)
|
||||
|
||||
rows, err := l.db.Query(sb.String(), args...)
|
||||
if err != nil {
|
||||
logStore("search: %v", err)
|
||||
return []Item{}
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
return scanItems(rows)
|
||||
}
|
||||
|
||||
// scanItems reads item rows into a slice.
|
||||
func scanItems(rows *sql.Rows) []Item {
|
||||
out := []Item{}
|
||||
for rows.Next() {
|
||||
var (
|
||||
it Item
|
||||
ts int64
|
||||
tickers string
|
||||
lat, lon sql.NullFloat64
|
||||
)
|
||||
if err := rows.Scan(&it.ID, &it.Kind, &it.Source, &ts, &it.Title, &it.Text,
|
||||
&tickers, &it.Country, &lat, &lon, &it.Payload); err != nil {
|
||||
logStore("scan: %v", err)
|
||||
continue
|
||||
}
|
||||
it.TS = time.Unix(ts, 0).UTC()
|
||||
it.Tickers = unpackTickers(tickers)
|
||||
if lat.Valid && lon.Valid {
|
||||
it.Lat, it.Lon, it.HasGeo = lat.Float64, lon.Float64, true
|
||||
}
|
||||
out = append(out, it)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// Count is one bucket of the analytics summary.
|
||||
type Count struct {
|
||||
Key string `json:"key"`
|
||||
Count int `json:"count"`
|
||||
}
|
||||
|
||||
// AnalyticsSummary is the cross-cutting "what's in the lake" view over the last
|
||||
// window: totals, breakdown by kind and source, and the top tickers seen.
|
||||
type AnalyticsSummary struct {
|
||||
WindowHours int `json:"windowHours"`
|
||||
Total int `json:"total"`
|
||||
ByKind []Count `json:"byKind"`
|
||||
BySource []Count `json:"bySource"`
|
||||
TopTickers []Count `json:"topTickers"`
|
||||
Since string `json:"since"`
|
||||
}
|
||||
|
||||
// Analytics summarizes the lake over the last `hours`. Degrades to zeros.
|
||||
func (l *Lake) Analytics(hours int) AnalyticsSummary {
|
||||
if hours <= 0 {
|
||||
hours = 24
|
||||
}
|
||||
out := AnalyticsSummary{WindowHours: hours, ByKind: []Count{}, BySource: []Count{}, TopTickers: []Count{}}
|
||||
if l == nil || l.db == nil {
|
||||
return out
|
||||
}
|
||||
since := time.Now().Add(-time.Duration(hours) * time.Hour)
|
||||
out.Since = since.UTC().Format(time.RFC3339)
|
||||
sinceUnix := since.Unix()
|
||||
|
||||
_ = l.db.QueryRow(`SELECT COUNT(*) FROM items WHERE ts>=?`, sinceUnix).Scan(&out.Total)
|
||||
out.ByKind = groupCount(l.db, `SELECT kind, COUNT(*) c FROM items WHERE ts>=? GROUP BY kind ORDER BY c DESC`, sinceUnix)
|
||||
out.BySource = groupCount(l.db, `SELECT source, COUNT(*) c FROM items WHERE ts>=? AND source!='' GROUP BY source ORDER BY c DESC LIMIT 15`, sinceUnix)
|
||||
out.TopTickers = topTickers(l.db, sinceUnix, 15)
|
||||
return out
|
||||
}
|
||||
|
||||
func groupCount(db *sql.DB, query string, args ...any) []Count {
|
||||
out := []Count{}
|
||||
rows, err := db.Query(query, args...)
|
||||
if err != nil {
|
||||
return out
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
for rows.Next() {
|
||||
var c Count
|
||||
if err := rows.Scan(&c.Key, &c.Count); err == nil {
|
||||
out = append(out, c)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// topTickers aggregates the packed tickers column in Go — cheap over a bounded
|
||||
// recent window and avoids a second normalized table for a summary stat.
|
||||
func topTickers(db *sql.DB, sinceUnix int64, n int) []Count {
|
||||
rows, err := db.Query(`SELECT tickers FROM items WHERE ts>=? AND tickers!='' LIMIT 20000`, sinceUnix)
|
||||
if err != nil {
|
||||
return []Count{}
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
freq := map[string]int{}
|
||||
for rows.Next() {
|
||||
var packed string
|
||||
if err := rows.Scan(&packed); err != nil {
|
||||
continue
|
||||
}
|
||||
for _, t := range unpackTickers(packed) {
|
||||
freq[t]++
|
||||
}
|
||||
}
|
||||
out := make([]Count, 0, len(freq))
|
||||
for k, v := range freq {
|
||||
out = append(out, Count{Key: k, Count: v})
|
||||
}
|
||||
sort.Slice(out, func(i, j int) bool {
|
||||
if out[i].Count != out[j].Count {
|
||||
return out[i].Count > out[j].Count
|
||||
}
|
||||
return out[i].Key < out[j].Key
|
||||
})
|
||||
if len(out) > n {
|
||||
out = out[:n]
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// Prune deletes items older than the retention window (by first-seen). The FTS
|
||||
// index is cleaned by the delete trigger. Returns rows removed.
|
||||
func (l *Lake) Prune(retention time.Duration) int64 {
|
||||
if l == nil || l.db == nil {
|
||||
return 0
|
||||
}
|
||||
if retention <= 0 {
|
||||
retention = DefaultRetention
|
||||
}
|
||||
cutoff := time.Now().Add(-retention).Unix()
|
||||
res, err := l.db.Exec(`DELETE FROM items WHERE created_at < ?`, cutoff)
|
||||
if err != nil {
|
||||
logStore("prune: %v", err)
|
||||
return 0
|
||||
}
|
||||
n, _ := res.RowsAffected()
|
||||
if n > 0 {
|
||||
logStore("pruned %d expired items", n)
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
// ── ticker packing ───────────────────────────────────────────────────────────
|
||||
|
||||
// packTickers stores tickers as a lowercase, space-padded token string
|
||||
// (" aapl btc ") so a token filter is an exact LIKE '% aapl %' — no false
|
||||
// substring hits, no separate table.
|
||||
func packTickers(tickers []string) string {
|
||||
if len(tickers) == 0 {
|
||||
return ""
|
||||
}
|
||||
seen := map[string]bool{}
|
||||
var b strings.Builder
|
||||
b.WriteByte(' ')
|
||||
for _, t := range tickers {
|
||||
t = strings.ToLower(strings.TrimSpace(t))
|
||||
if t == "" || seen[t] {
|
||||
continue
|
||||
}
|
||||
seen[t] = true
|
||||
b.WriteString(t)
|
||||
b.WriteByte(' ')
|
||||
}
|
||||
if b.Len() == 1 {
|
||||
return ""
|
||||
}
|
||||
return b.String()
|
||||
}
|
||||
|
||||
func unpackTickers(packed string) []string {
|
||||
f := strings.Fields(packed)
|
||||
if len(f) == 0 {
|
||||
return nil
|
||||
}
|
||||
return f
|
||||
}
|
||||
|
||||
// ftsQuery turns free user text into a SAFE FTS5 MATCH expression: alphanumeric
|
||||
// tokens, each double-quoted (so FTS control chars can never cause a syntax
|
||||
// error), AND-ed together. Empty when there are no usable tokens (caller then
|
||||
// browses by recency). This is the SSRF-equivalent boundary for FTS — untrusted
|
||||
// input can never reach the query grammar unescaped.
|
||||
func ftsQuery(q string) string {
|
||||
var tokens []string
|
||||
for _, f := range strings.FieldsFunc(q, func(r rune) bool {
|
||||
return !(r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9')
|
||||
}) {
|
||||
if len(tokens) >= 12 {
|
||||
break
|
||||
}
|
||||
tokens = append(tokens, `"`+f+`"`)
|
||||
}
|
||||
return strings.Join(tokens, " ")
|
||||
}
|
||||
@@ -0,0 +1,161 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestLakeIngestSearchRoundTrip(t *testing.T) {
|
||||
db := openTestDB(t)
|
||||
now := time.Now()
|
||||
items := []Item{
|
||||
{ID: "n1", Kind: "news", Source: "coindesk.com", TS: now, Title: "Bitcoin ETF approved by regulators", Tickers: []string{"BTC"}},
|
||||
{ID: "n2", Kind: "news", Source: "sec.gov", TS: now, Title: "SEC announces new disclosure rules"},
|
||||
{ID: "o1", Kind: "observation", Source: "gdelt", TS: now, Title: "Ukraine", Country: "UA"},
|
||||
}
|
||||
if err := db.Lake.insert(items); err != nil {
|
||||
t.Fatalf("insert: %v", err)
|
||||
}
|
||||
|
||||
// Full-text: "regulators" hits only n1.
|
||||
if got := db.Lake.Search(SearchQuery{Q: "regulators"}); len(got) != 1 || got[0].ID != "n1" {
|
||||
t.Fatalf("search regulators = %+v, want [n1]", ids(got))
|
||||
}
|
||||
// Empty query browses by recency across ALL kinds.
|
||||
if got := db.Lake.Search(SearchQuery{}); len(got) != 3 {
|
||||
t.Fatalf("browse all = %v, want 3 rows", ids(got))
|
||||
}
|
||||
// Tickers survive the round-trip.
|
||||
got := db.Lake.Search(SearchQuery{Q: "bitcoin"})
|
||||
if len(got) != 1 || len(got[0].Tickers) != 1 || got[0].Tickers[0] != "btc" {
|
||||
t.Fatalf("tickers round-trip = %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSearchFTSRanking(t *testing.T) {
|
||||
db := openTestDB(t)
|
||||
now := time.Now()
|
||||
// a mentions the term twice in the title, b once in the body → a ranks first.
|
||||
if err := db.Lake.insert([]Item{
|
||||
{ID: "a", Kind: "news", TS: now, Title: "Bitcoin Bitcoin rally continues"},
|
||||
{ID: "b", Kind: "news", TS: now, Title: "Markets update", Text: "a passing bitcoin mention"},
|
||||
}); err != nil {
|
||||
t.Fatalf("insert: %v", err)
|
||||
}
|
||||
got := db.Lake.Search(SearchQuery{Q: "bitcoin"})
|
||||
if len(got) != 2 {
|
||||
t.Fatalf("want 2 hits, got %v", ids(got))
|
||||
}
|
||||
if got[0].ID != "a" {
|
||||
t.Fatalf("bm25 ranking = %v, want a ranked first", ids(got))
|
||||
}
|
||||
}
|
||||
|
||||
func TestSearchFilters(t *testing.T) {
|
||||
db := openTestDB(t)
|
||||
now := time.Now()
|
||||
old := now.Add(-48 * time.Hour)
|
||||
if err := db.Lake.insert([]Item{
|
||||
{ID: "fresh", Kind: "news", Source: "cnbc.com", TS: now, Title: "Fed rate decision", Country: "US", Tickers: []string{"SPY"}},
|
||||
{ID: "stale", Kind: "news", Source: "cnbc.com", TS: old, Title: "Fed rate decision old", Country: "US"},
|
||||
{ID: "obs", Kind: "observation", TS: now, Title: "Fed watch", Country: "US"},
|
||||
}); err != nil {
|
||||
t.Fatalf("insert: %v", err)
|
||||
}
|
||||
// kind filter
|
||||
if got := db.Lake.Search(SearchQuery{Q: "fed", Kind: "observation"}); len(got) != 1 || got[0].ID != "obs" {
|
||||
t.Fatalf("kind filter = %v", ids(got))
|
||||
}
|
||||
// since filter drops the 48h-old row
|
||||
if got := db.Lake.Search(SearchQuery{Q: "fed", Kind: "news", Since: now.Add(-24 * time.Hour)}); len(got) != 1 || got[0].ID != "fresh" {
|
||||
t.Fatalf("since filter = %v", ids(got))
|
||||
}
|
||||
// ticker filter (exact token, no substring false-hit)
|
||||
if got := db.Lake.Search(SearchQuery{Ticker: "SPY"}); len(got) != 1 || got[0].ID != "fresh" {
|
||||
t.Fatalf("ticker filter = %v", ids(got))
|
||||
}
|
||||
if got := db.Lake.Search(SearchQuery{Ticker: "SP"}); len(got) != 0 {
|
||||
t.Fatalf("ticker substring must not match, got %v", ids(got))
|
||||
}
|
||||
// country filter
|
||||
if got := db.Lake.Search(SearchQuery{Country: "US"}); len(got) != 3 {
|
||||
t.Fatalf("country filter = %v, want all 3 US rows", ids(got))
|
||||
}
|
||||
}
|
||||
|
||||
func TestRetentionPrune(t *testing.T) {
|
||||
db := openTestDB(t)
|
||||
now := time.Now()
|
||||
// Insert one expired and one fresh row with explicit created_at (white-box).
|
||||
mustExec(t, db, `INSERT INTO items(id,kind,source,ts,title,text,tickers,country,lat,lon,payload,created_at)
|
||||
VALUES ('old','news','s',?, 'old title','','','',NULL,NULL,'', ?)`,
|
||||
now.Unix(), now.Add(-10*24*time.Hour).Unix())
|
||||
mustExec(t, db, `INSERT INTO items(id,kind,source,ts,title,text,tickers,country,lat,lon,payload,created_at)
|
||||
VALUES ('new','news','s',?, 'new title','','','',NULL,NULL,'', ?)`,
|
||||
now.Unix(), now.Unix())
|
||||
|
||||
if n := db.Lake.Prune(7 * 24 * time.Hour); n != 1 {
|
||||
t.Fatalf("prune removed %d, want 1", n)
|
||||
}
|
||||
// The FTS index must follow the delete (trigger) — the pruned row is unsearchable.
|
||||
if got := db.Lake.Search(SearchQuery{Q: "old"}); len(got) != 0 {
|
||||
t.Fatalf("pruned row still searchable via FTS: %v", ids(got))
|
||||
}
|
||||
if got := db.Lake.Search(SearchQuery{Q: "new"}); len(got) != 1 {
|
||||
t.Fatalf("fresh row lost after prune: %v", ids(got))
|
||||
}
|
||||
}
|
||||
|
||||
func TestAnalyticsSummary(t *testing.T) {
|
||||
db := openTestDB(t)
|
||||
now := time.Now()
|
||||
if err := db.Lake.insert([]Item{
|
||||
{ID: "1", Kind: "news", Source: "coindesk.com", TS: now, Title: "a", Tickers: []string{"BTC", "ETH"}},
|
||||
{ID: "2", Kind: "news", Source: "coindesk.com", TS: now, Title: "b", Tickers: []string{"BTC"}},
|
||||
{ID: "3", Kind: "observation", Source: "gdelt", TS: now, Title: "c"},
|
||||
}); err != nil {
|
||||
t.Fatalf("insert: %v", err)
|
||||
}
|
||||
a := db.Lake.Analytics(24)
|
||||
if a.Total != 3 {
|
||||
t.Fatalf("total = %d, want 3", a.Total)
|
||||
}
|
||||
if kind := countFor(a.ByKind, "news"); kind != 2 {
|
||||
t.Fatalf("byKind news = %d, want 2", kind)
|
||||
}
|
||||
if src := countFor(a.BySource, "coindesk.com"); src != 2 {
|
||||
t.Fatalf("bySource coindesk = %d, want 2", src)
|
||||
}
|
||||
if btc := countFor(a.TopTickers, "btc"); btc != 2 {
|
||||
t.Fatalf("topTickers btc = %d, want 2", btc)
|
||||
}
|
||||
if len(a.TopTickers) == 0 || a.TopTickers[0].Key != "btc" {
|
||||
t.Fatalf("top ticker = %+v, want btc first", a.TopTickers)
|
||||
}
|
||||
}
|
||||
|
||||
// ── helpers ──────────────────────────────────────────────────────────────────
|
||||
|
||||
func ids(items []Item) []string {
|
||||
out := make([]string, len(items))
|
||||
for i, it := range items {
|
||||
out[i] = it.ID
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func countFor(cs []Count, key string) int {
|
||||
for _, c := range cs {
|
||||
if c.Key == key {
|
||||
return c.Count
|
||||
}
|
||||
}
|
||||
return -1
|
||||
}
|
||||
|
||||
func mustExec(t *testing.T, db *DB, query string, args ...any) {
|
||||
t.Helper()
|
||||
if _, err := db.sql.Exec(query, args...); err != nil {
|
||||
t.Fatalf("exec %q: %v", query, err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,77 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Settings persists a per-identity JSON blob (dashboard layout, panel geometry,
|
||||
// preferences) SERVER-SIDE, keyed by (org, user_sub, project) from the IAM
|
||||
// bearer, so a signed-in user's dashboard syncs across devices. It is the "one
|
||||
// driver" companion to the lake — same SQLite DB, its own table, orthogonal
|
||||
// concern. Anonymous callers never reach here (the handler gates on the bearer);
|
||||
// they keep using localStorage.
|
||||
type Settings struct {
|
||||
db *sql.DB // nil in degraded mode
|
||||
}
|
||||
|
||||
// Identity keys a settings row. Project defaults to "default" upstream so a
|
||||
// user's single dashboard has a stable key.
|
||||
type Identity struct {
|
||||
Org string
|
||||
UserSub string
|
||||
Project string
|
||||
}
|
||||
|
||||
func (id Identity) normalize() Identity {
|
||||
id.Org = strings.TrimSpace(id.Org)
|
||||
id.UserSub = strings.TrimSpace(id.UserSub)
|
||||
id.Project = strings.TrimSpace(id.Project)
|
||||
if id.Project == "" {
|
||||
id.Project = "default"
|
||||
}
|
||||
return id
|
||||
}
|
||||
|
||||
// Get returns the stored settings blob for an identity, or (nil,false) when
|
||||
// absent or degraded. The blob is returned verbatim (already valid JSON).
|
||||
func (s *Settings) Get(id Identity) (json.RawMessage, bool) {
|
||||
if s == nil || s.db == nil {
|
||||
return nil, false
|
||||
}
|
||||
id = id.normalize()
|
||||
if id.UserSub == "" {
|
||||
return nil, false
|
||||
}
|
||||
var blob string
|
||||
err := s.db.QueryRow(
|
||||
`SELECT blob FROM settings WHERE org=? AND user_sub=? AND project=?`,
|
||||
id.Org, id.UserSub, id.Project).Scan(&blob)
|
||||
if err != nil {
|
||||
return nil, false
|
||||
}
|
||||
return json.RawMessage(blob), true
|
||||
}
|
||||
|
||||
// Put upserts an identity's settings blob. blob must be valid JSON (validated by
|
||||
// the caller at the boundary). Reports whether it was stored (false = degraded).
|
||||
func (s *Settings) Put(id Identity, blob json.RawMessage) bool {
|
||||
if s == nil || s.db == nil {
|
||||
return false
|
||||
}
|
||||
id = id.normalize()
|
||||
if id.UserSub == "" {
|
||||
return false
|
||||
}
|
||||
_, err := s.db.Exec(`INSERT INTO settings (org, user_sub, project, blob, updated_at)
|
||||
VALUES (?, ?, ?, ?, ?)
|
||||
ON CONFLICT(org, user_sub, project) DO UPDATE SET blob=excluded.blob, updated_at=excluded.updated_at`,
|
||||
id.Org, id.UserSub, id.Project, string(blob), time.Now().Unix())
|
||||
if err != nil {
|
||||
logStore("settings put: %v", err)
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
@@ -0,0 +1,73 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestSettingsUpsertAndIdentityIsolation(t *testing.T) {
|
||||
db := openTestDB(t)
|
||||
alice := Identity{Org: "acme", UserSub: "alice", Project: "default"}
|
||||
bob := Identity{Org: "acme", UserSub: "bob", Project: "default"}
|
||||
|
||||
if !db.Settings.Put(alice, json.RawMessage(`{"layout":"grid"}`)) {
|
||||
t.Fatal("put alice failed")
|
||||
}
|
||||
if !db.Settings.Put(bob, json.RawMessage(`{"layout":"list"}`)) {
|
||||
t.Fatal("put bob failed")
|
||||
}
|
||||
|
||||
// Each identity reads back its OWN blob — no cross-talk.
|
||||
if got, ok := db.Settings.Get(alice); !ok || string(got) != `{"layout":"grid"}` {
|
||||
t.Fatalf("get alice = %q ok=%v", got, ok)
|
||||
}
|
||||
if got, ok := db.Settings.Get(bob); !ok || string(got) != `{"layout":"list"}` {
|
||||
t.Fatalf("get bob = %q ok=%v", got, ok)
|
||||
}
|
||||
|
||||
// Upsert replaces alice's blob and leaves bob untouched.
|
||||
if !db.Settings.Put(alice, json.RawMessage(`{"layout":"masonry","cell":42}`)) {
|
||||
t.Fatal("upsert alice failed")
|
||||
}
|
||||
if got, _ := db.Settings.Get(alice); string(got) != `{"layout":"masonry","cell":42}` {
|
||||
t.Fatalf("upsert alice = %q", got)
|
||||
}
|
||||
if got, _ := db.Settings.Get(bob); string(got) != `{"layout":"list"}` {
|
||||
t.Fatalf("bob mutated by alice upsert: %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSettingsProjectIsolationAndDefault(t *testing.T) {
|
||||
db := openTestDB(t)
|
||||
base := Identity{Org: "acme", UserSub: "alice"} // Project "" → "default"
|
||||
proj := Identity{Org: "acme", UserSub: "alice", Project: "trading"}
|
||||
|
||||
if !db.Settings.Put(base, json.RawMessage(`{"p":"default"}`)) {
|
||||
t.Fatal("put default failed")
|
||||
}
|
||||
if !db.Settings.Put(proj, json.RawMessage(`{"p":"trading"}`)) {
|
||||
t.Fatal("put trading failed")
|
||||
}
|
||||
// Same user, different project → different blobs.
|
||||
if got, _ := db.Settings.Get(Identity{Org: "acme", UserSub: "alice", Project: "default"}); string(got) != `{"p":"default"}` {
|
||||
t.Fatalf("default project = %q", got)
|
||||
}
|
||||
if got, _ := db.Settings.Get(proj); string(got) != `{"p":"trading"}` {
|
||||
t.Fatalf("trading project = %q", got)
|
||||
}
|
||||
// Empty project normalizes to "default".
|
||||
if got, _ := db.Settings.Get(base); string(got) != `{"p":"default"}` {
|
||||
t.Fatalf("empty project did not normalize to default: %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSettingsMissingAndNoSub(t *testing.T) {
|
||||
db := openTestDB(t)
|
||||
if _, ok := db.Settings.Get(Identity{Org: "acme", UserSub: "ghost"}); ok {
|
||||
t.Fatal("get for absent identity reported ok")
|
||||
}
|
||||
// A missing subject is not a valid identity — never store or read.
|
||||
if db.Settings.Put(Identity{Org: "acme", UserSub: ""}, json.RawMessage(`{}`)) {
|
||||
t.Fatal("put with empty sub reported stored")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,139 @@
|
||||
// Package store is world's embedded, CGO-free datastore: the queryable
|
||||
// ingested-data LAKE (full-text searchable) and per-identity SETTINGS, both in
|
||||
// ONE SQLite database (modernc.org/sqlite — pure Go, FTS5 built in).
|
||||
//
|
||||
// It is decomplected from HTTP and from the shared feed hot-cache: this package
|
||||
// only holds durable, queryable rows and answers value queries. The instant
|
||||
// feed body cache lives in package kv (hanzo-kv, shared across pods); SQLite is
|
||||
// the "one place to query everything" — search, analytics, and signed-in
|
||||
// settings. One driver, two logical stores.
|
||||
//
|
||||
// Never-5xx: if the database cannot be opened the returned *DB is still usable —
|
||||
// every method degrades to an empty/no-op result rather than failing. A single
|
||||
// serialized connection (SetMaxOpenConns(1)) makes all access deterministic and
|
||||
// free of SQLITE_BUSY, which is correct for a per-pod cache/lake where every
|
||||
// operation is sub-millisecond.
|
||||
package store
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"log"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"time"
|
||||
|
||||
_ "modernc.org/sqlite"
|
||||
)
|
||||
|
||||
// DB owns the embedded SQLite handle and the two logical stores layered on it.
|
||||
// sql is nil only when the database could not be opened (degraded mode).
|
||||
type DB struct {
|
||||
sql *sql.DB
|
||||
Lake *Lake
|
||||
Settings *Settings
|
||||
}
|
||||
|
||||
// dbFile is the single embedded database filename under the data dir.
|
||||
const dbFile = "world.db"
|
||||
|
||||
// schema is the full DDL, applied idempotently on Open. The lake is an external
|
||||
// content FTS5 index (title+text) kept in sync by triggers, so bm25 ranking and
|
||||
// prune-driven deletes stay correct with no manual FTS bookkeeping.
|
||||
var schema = []string{
|
||||
`CREATE TABLE IF NOT EXISTS items (
|
||||
id TEXT PRIMARY KEY,
|
||||
kind TEXT NOT NULL,
|
||||
source TEXT NOT NULL DEFAULT '',
|
||||
ts INTEGER NOT NULL, -- item's own time (unix seconds)
|
||||
title TEXT NOT NULL DEFAULT '',
|
||||
text TEXT NOT NULL DEFAULT '',
|
||||
tickers TEXT NOT NULL DEFAULT '', -- space-delimited, lowercase, padded ' a b '
|
||||
country TEXT NOT NULL DEFAULT '',
|
||||
lat REAL,
|
||||
lon REAL,
|
||||
payload TEXT NOT NULL DEFAULT '', -- compact original item JSON
|
||||
created_at INTEGER NOT NULL -- first-seen (unix seconds); drives retention
|
||||
)`,
|
||||
`CREATE INDEX IF NOT EXISTS items_kind_ts ON items(kind, ts)`,
|
||||
`CREATE INDEX IF NOT EXISTS items_ts ON items(ts)`,
|
||||
`CREATE INDEX IF NOT EXISTS items_created ON items(created_at)`,
|
||||
`CREATE INDEX IF NOT EXISTS items_country ON items(country)`,
|
||||
`CREATE VIRTUAL TABLE IF NOT EXISTS items_fts USING fts5(
|
||||
title, text, content='items', content_rowid='rowid'
|
||||
)`,
|
||||
`CREATE TRIGGER IF NOT EXISTS items_ai AFTER INSERT ON items BEGIN
|
||||
INSERT INTO items_fts(rowid, title, text) VALUES (new.rowid, new.title, new.text);
|
||||
END`,
|
||||
`CREATE TRIGGER IF NOT EXISTS items_ad AFTER DELETE ON items BEGIN
|
||||
INSERT INTO items_fts(items_fts, rowid, title, text) VALUES ('delete', old.rowid, old.title, old.text);
|
||||
END`,
|
||||
`CREATE TRIGGER IF NOT EXISTS items_au AFTER UPDATE ON items BEGIN
|
||||
INSERT INTO items_fts(items_fts, rowid, title, text) VALUES ('delete', old.rowid, old.title, old.text);
|
||||
INSERT INTO items_fts(rowid, title, text) VALUES (new.rowid, new.title, new.text);
|
||||
END`,
|
||||
`CREATE TABLE IF NOT EXISTS settings (
|
||||
org TEXT NOT NULL,
|
||||
user_sub TEXT NOT NULL,
|
||||
project TEXT NOT NULL,
|
||||
blob TEXT NOT NULL,
|
||||
updated_at INTEGER NOT NULL,
|
||||
PRIMARY KEY (org, user_sub, project)
|
||||
)`,
|
||||
}
|
||||
|
||||
// DefaultRetention is how long ingested items are kept before the prune job
|
||||
// deletes them (rolling window). Overridable via the caller.
|
||||
const DefaultRetention = 7 * 24 * time.Hour
|
||||
|
||||
// Open opens (creating if needed) the embedded database under dir and applies
|
||||
// the schema. It ALWAYS returns a usable *DB: on any failure it logs, returns a
|
||||
// degraded DB (empty results, no-op writes) plus the error, so callers never
|
||||
// have to nil-check and the service never 5xxes over storage.
|
||||
func Open(dir string, retention time.Duration) (*DB, error) {
|
||||
if retention <= 0 {
|
||||
retention = DefaultRetention
|
||||
}
|
||||
degraded := &DB{Lake: newLake(nil, retention), Settings: &Settings{}}
|
||||
|
||||
if err := os.MkdirAll(dir, 0o755); err != nil {
|
||||
return degraded, err
|
||||
}
|
||||
dsn := "file:" + filepath.Join(dir, dbFile) +
|
||||
"?_pragma=journal_mode(WAL)&_pragma=busy_timeout(5000)&_pragma=synchronous(NORMAL)&_pragma=foreign_keys(ON)"
|
||||
sqldb, err := sql.Open("sqlite", dsn)
|
||||
if err != nil {
|
||||
return degraded, err
|
||||
}
|
||||
// One connection: all access serialized in-process — deterministic, no
|
||||
// SQLITE_BUSY. Every op is sub-ms so serialization costs nothing here.
|
||||
sqldb.SetMaxOpenConns(1)
|
||||
if err := sqldb.Ping(); err != nil {
|
||||
_ = sqldb.Close()
|
||||
return degraded, err
|
||||
}
|
||||
for _, ddl := range schema {
|
||||
if _, err := sqldb.Exec(ddl); err != nil {
|
||||
_ = sqldb.Close()
|
||||
return degraded, err
|
||||
}
|
||||
}
|
||||
return &DB{
|
||||
sql: sqldb,
|
||||
Lake: newLake(sqldb, retention),
|
||||
Settings: &Settings{db: sqldb},
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Enabled reports whether the database opened successfully (durable mode).
|
||||
func (d *DB) Enabled() bool { return d != nil && d.sql != nil }
|
||||
|
||||
// Close flushes and closes the database. Safe on a degraded DB.
|
||||
func (d *DB) Close() error {
|
||||
if d == nil || d.sql == nil {
|
||||
return nil
|
||||
}
|
||||
return d.sql.Close()
|
||||
}
|
||||
|
||||
// logStore namespaces the store's best-effort warnings.
|
||||
func logStore(format string, args ...any) { log.Printf("world-store: "+format, args...) }
|
||||
@@ -0,0 +1,73 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// openTestDB opens a fresh embedded datastore in a temp dir. The mere fact this
|
||||
// succeeds under `CGO_ENABLED=0 go test` is the CGO-free assertion: modernc's
|
||||
// pure-Go SQLite (with FTS5) links and runs without a C toolchain.
|
||||
func openTestDB(t *testing.T) *DB {
|
||||
t.Helper()
|
||||
db, err := Open(t.TempDir(), time.Hour)
|
||||
if err != nil {
|
||||
t.Fatalf("Open: %v", err)
|
||||
}
|
||||
if !db.Enabled() {
|
||||
t.Fatal("Open returned a disabled DB")
|
||||
}
|
||||
t.Cleanup(func() { _ = db.Close() })
|
||||
return db
|
||||
}
|
||||
|
||||
func TestOpenAppliesSchemaAndFTS5(t *testing.T) {
|
||||
db := openTestDB(t)
|
||||
// FTS5 must be compiled into modernc's SQLite — a MATCH query over the virtual
|
||||
// table would error otherwise. Ingest one row and match it.
|
||||
if err := db.Lake.insert([]Item{{ID: "x", Kind: "news", Title: "Bitcoin surges today"}}); err != nil {
|
||||
t.Fatalf("insert: %v", err)
|
||||
}
|
||||
got := db.Lake.Search(SearchQuery{Q: "bitcoin"})
|
||||
if len(got) != 1 || got[0].ID != "x" {
|
||||
t.Fatalf("FTS5 search = %+v, want the one bitcoin row", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOpenDegradedIsUsableNeverNil(t *testing.T) {
|
||||
// A path under a regular file cannot be a directory → Open fails, but must
|
||||
// still return a usable degraded DB (never-5xx contract at the storage layer).
|
||||
f := filepath.Join(t.TempDir(), "not-a-dir")
|
||||
if err := os.WriteFile(f, []byte("x"), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
db, err := Open(filepath.Join(f, "sub"), time.Hour)
|
||||
if err == nil {
|
||||
t.Fatal("expected an error opening under a file")
|
||||
}
|
||||
if db == nil {
|
||||
t.Fatal("Open returned nil DB on error")
|
||||
}
|
||||
if db.Enabled() {
|
||||
t.Fatal("degraded DB reports Enabled")
|
||||
}
|
||||
// Every method must no-op / empty rather than panic.
|
||||
if got := db.Lake.Search(SearchQuery{Q: "x"}); len(got) != 0 {
|
||||
t.Fatalf("degraded Search = %v, want empty", got)
|
||||
}
|
||||
db.Lake.Add(Item{ID: "a", Kind: "news", Title: "t"})
|
||||
if n := db.Lake.Prune(time.Hour); n != 0 {
|
||||
t.Fatalf("degraded Prune = %d, want 0", n)
|
||||
}
|
||||
if _, ok := db.Settings.Get(Identity{UserSub: "u"}); ok {
|
||||
t.Fatal("degraded Settings.Get returned ok")
|
||||
}
|
||||
if db.Settings.Put(Identity{UserSub: "u"}, []byte(`{}`)) {
|
||||
t.Fatal("degraded Settings.Put reported stored")
|
||||
}
|
||||
if a := db.Lake.Analytics(24); a.Total != 0 {
|
||||
t.Fatalf("degraded Analytics total = %d, want 0", a.Total)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user