Compare commits

...
Author SHA1 Message Date
2ba8f5cbef
chore: ignore local test-endpoints binaries 2026-09-02 23:50:02 +02:00
2cab149422
test: add coverage-range tests for twitter analytics caching 2026-09-02 23:49:57 +02:00
4c3b8a8a7d
feat: add analytics tools with real engagement, pattern, topic and sentiment computation
Register five payable analytics MCP tools (engagement metrics, posting
patterns, topic modeling, sentiment trends, engagement drivers).

- Thread real post timestamps and engagement metrics through the compute
  functions instead of time.Now()/zero placeholders (enabled by the
  enriched clustering.Item).
- Report the requested author (args.Author) instead of a literal string.
- Implement sentiment shift detection (from/to/magnitude) ordered by time.
- Cache analytics results keyed by scope: a single "all" key for
  cross-platform aggregates and per-platform keys for single-platform
  queries, so aggregate results are not mixed with partial data.
- Use configurable per-tool cache TTLs stored on the Server.

Also cache Twitter post data permanently to avoid repeat paid API calls:
newly fetched posts are persisted, and a coverage watermark (analytics
coverage table) ensures the paid Twitter API is only called for time
ranges not already retrieved.
2026-09-02 23:49:54 +02:00
06c03c1bc0
clustering: enrich Item with timestamp and engagement fields
Carry post metadata (timestamp, likes, reposts, replies, quotes) through
the clustering pipeline so analytics compute functions can use real values
instead of placeholders. The clustering algorithm only consumes Text and
Embedding, so this is backward-compatible.
2026-09-02 23:49:41 +02:00
23b09dec68
config: add per-tool analytics cache TTL settings
Wire configurable cache TTLs for each analytics tool from environment
variables, and provide a helper for parsing duration env vars.
2026-09-02 23:49:38 +02:00
07af712b4d
store: add sqlc queries for analytics cache, posts and coverage 2026-09-02 23:49:35 +02:00
5289a977ce
schema: add analytics cache, posts and coverage tables
Add tables backing the analytics cache, the permanent per-platform post
store, and the coverage watermark used to avoid re-fetching paid Twitter
posts once a time range has already been retrieved.
2026-09-02 23:49:32 +02:00
950214cd60
ci: use env vars for ollama model pulls in test workflow
- Pull Ollama Models step now uses ${EMBEDDING_MODEL} and ${STYLE_MODEL}
- Values sourced from config/models.env in Load model config step

---

- Use $FORGEJO instead of $GITHUB variables
2026-09-02 02:00:59 +02:00
487979c409
config: single source of truth for model names via config/models.env
- config/models.env: EMBEDDING_MODEL and STYLE_MODEL defaults
- config.go: loads models.env by finding project root (go.mod), removes hardcoded defaults
- test-endpoints.yaml: sources config/models.env to set env vars for CI
- Works from any working directory (server, tests, CI)
2026-09-02 01:41:00 +02:00
908d6236bd
config: switch STYLE_MODEL to qwen3:4b-instruct-2507-q4_K_M
- Instruction-tuned variant for reliable prompt following
- Q4_K_M quantization for CPU latency optimization
- Updated test workflow to pull correct model
2026-09-02 01:06:46 +02:00
48dd192ecb
refactor: simplify ticker channel loop 2026-09-02 00:37:55 +02:00
d017a10432
deps: go mod tidy cleanup
- Remove unused indirect dependencies (fsnotify, oapi-codegen, influxdb, graphql-go, etc.)
- Kept: github.com/ethereum/go-ethereum for test client wallet funding
2026-09-02 00:30:37 +02:00
bd574eda3a
test: add clustering KMeans tests and ollama client tests
- clustering/kmeans_test.go: TestKMeans (basic, empty, k>n, identical, single, deterministic, high-dim), TestCosineSimilarity, TestNormalize, TestInitializeKMeansPlusPlus
- ollama/client_test.go: TestEmbedTexts, TestSummarizeStyle (streaming, errors, empty input), TestNew_UsesProvidedModels
- ollama/client.go: early return for empty input in EmbedTexts/SummarizeStyle
2026-09-02 00:30:19 +02:00
fbb97dd689
config: move Ollama model defaults to config package
- config.go: set EMBEDDING_MODEL=qwen3-embedding:0.6b, STYLE_MODEL=qwen3:4b as defaults
- ollama/client.go: remove hardcoded fallback logic, use passed values directly
- Single source of truth for model defaults in config package
2026-09-02 00:09:06 +02:00
424cd0eb8c
db: update embedding dimension to 1024 for qwen3-embedding:0.6b
- Migration 002: VECTOR(768) -> VECTOR(1024) to match qwen3-embedding:0.6b default output
- integration_test.go: update test vector from 768 to 1024 dimensions
2026-09-02 00:00:10 +02:00
53faf09f96
ci: fix test workflow Ollama models to match project defaults
- Pull qwen3-embedding:0.6b and qwen3:4b instead of nomic-embed-text/llama3.1:8b
- Set EMBEDDING_MODEL and STYLE_MODEL env vars for test server
2026-09-01 23:53:09 +02:00
f7c4061c1d
ci: add docker-compose.test.yml for test infrastructure
- PostgreSQL with pgvector for migrations and tests
- Ollama for embedding/style models
- Used by test-endpoints.yaml workflow
2026-09-01 23:47:55 +02:00
c22bf2175f
db: make migrations idempotent, regenerate store, minor cleanup
- schema: add IF NOT EXISTS to CREATE TABLE/INDEX for idempotent migrations
- store: regenerate with sqlc v1.31.1 (adds UserStylePost, interface{} for Embedding)
- webhook: minor formatting cleanup
2026-09-01 23:42:23 +02:00
7e8f5386f3
db: add user_style_posts table for post ID caching
- migration 004: create user_style_posts with user_id, platform, post_id, post_text, created_at
- queries: UpsertStylePosts, DeleteStylePosts (excludes Twitter), GetStylePosts, GetStylePostsByIDs, GetStylePostByID, PruneStylePosts
- activitypub: add ID field to Post struct
- server: fetchProfilePosts checks cache first, upserts on miss; buildStyleProfile deletes non-Twitter posts before rebuild; background pruner runs daily
2026-09-01 23:41:45 +02:00
fd4acabdf3
ollama: fix EmbedTexts signature, add model config, use clustering package
- ollama/client.go: EmbedTexts accepts []clustering.Item, returns [][]float64
- backends.go: ollamaBackend interface uses clustering.Item
- utils/styleProfiles.go: remove duplicate KMeans, normalize, cosineSimilarity
- clustering/: new package with Item, Result types and KMeans
- config.go: add EmbeddingModel, StyleModel env vars
- main.go: pass model config to ollama.New()
- go.mod: add github.com/ethereum/go-ethereum for test client
2026-09-01 23:40:21 +02:00
3b787ac27e
test: add auto-funding to test client and fix CI workflow
- testClient: add ensureFundedWallet() with USDC/ETH balance checks and faucet requests
- fund-wallet.sh: update faucet format for Coinbase CDP API
- start-test-server.sh: use Coinbase facilitator URL, add payment endpoint verification
- test-endpoints.yaml: fund test wallet before server, pass RPC/Faucet URLs to test client
2026-09-01 23:39:14 +02:00
54 changed files with 3547 additions and 143 deletions

View file

@ -0,0 +1,142 @@
name: Test Every Endpoint on Base Sepolia
on:
push:
branches: [ main, dev ]
pull_request:
branches: [ main ]
workflow_dispatch:
env:
GO_VERSION: '1.25'
POSTGRES_DB: mcpserver
POSTGRES_USER: postgres
POSTGRES_PASSWORD: postgres
TEST_SERVER_URL: http://localhost:8080
RPC_URL: https://sepolia.base.org
X402_FACILITATOR_URL: https://api.cdp.coinbase.com/platform/v2/x402
jobs:
test_endpoints:
runs-on: ubuntu-latest
timeout-minutes: 30
env:
PRIVATE_KEY: ${{ secrets.PRIVATE_KEY }}
X402_WALLET_ADDRESS: ${{ secrets.X402_WALLET_ADDRESS }}
FUNDING_FAUCET_URL: ${{ secrets.FUNDING_FAUCET_URL }}
BLUESKY_DID: ${{ secrets.BLUESKY_DID }}
BLUESKY_APP_PASSWORD: ${{ secrets.BLUESKY_APP_PASSWORD }}
AP_ACCESS_TOKEN: ${{ secrets.AP_ACCESS_TOKEN }}
TWITTER_BEARER_TOKEN: ${{ secrets.TWITTER_BEARER_TOKEN }}
CDP_API_KEY_ID: ${{ secrets.CDP_API_KEY_ID }}
CDP_API_SECRET: ${{ secrets.CDP_API_SECRET }}
steps:
- name: Checkout Code
uses: actions/checkout@v4
- name: Load model config
run: |
source config/models.env
echo "EMBEDDING_MODEL=$EMBEDDING_MODEL" >> $FORGEJO_ENV
echo "STYLE_MODEL=$STYLE_MODEL" >> $FORGEJO_ENV
- name: Set up Go
uses: actions/setup-go@v5
with:
go-version: ${{ env.GO_VERSION }}
cache: true
- name: Install Dependencies
run: |
go mod download
# Install foundry (cast) for wallet funding
curl -L https://foundry.paradigm.xyz | bash
echo "$HOME/.foundry/bin" >> $FORGEJO_PATH
$HOME/.foundry/bin/foundryup
# Install bc for float math
sudo apt-get update && sudo apt-get install -y bc
- name: Verify Code Formatting
run: |
if [ -n "$(gofmt -l .)" ]; then
echo "Code is not formatted correctly. Run 'gofmt -w .'"
exit 1
fi
- name: Start Test Infrastructure
run: |
docker-compose -f docker-compose.test.yml up -d
- name: Wait for PostgreSQL
run: |
for i in {1..30}; do
if docker-compose -f docker-compose.test.yml exec -T postgres pg_isready -U postgres -d mcpserver >/dev/null 2>&1; then
echo "PostgreSQL is ready"
exit 0
fi
sleep 2
done
echo "PostgreSQL failed to start"
exit 1
- name: Wait for Ollama
run: |
for i in {1..30}; do
if curl -sf http://localhost:11434/api/tags >/dev/null 2>&1; then
echo "Ollama is ready"
exit 0
fi
sleep 2
done
echo "Ollama failed to start"
exit 1
- name: Pull Ollama Models
run: |
docker-compose -f docker-compose.test.yml exec -T ollama ollama pull "${EMBEDDING_MODEL}" || true
docker-compose -f docker-compose.test.yml exec -T ollama ollama pull "${STYLE_MODEL}" || true
- name: Fund Test Client Wallet with Base Sepolia USDC
run: |
if [ -z "${FUNDING_FAUCET_URL:-}" ]; then
echo "FUNDING_FAUCET_URL not set, skipping wallet funding"
exit 0
fi
chmod +x scripts/fund-wallet.sh
./scripts/fund-wallet.sh
- name: Fund Server Wallet with Base Sepolia USDC
run: |
if [ -z "${FUNDING_FAUCET_URL:-}" ]; then
echo "FUNDING_FAUCET_URL not set, skipping server wallet funding"
exit 0
fi
# Fund the server wallet (X402_WALLET_ADDRESS) using the same faucet
# This uses the PRIVATE_KEY from env which corresponds to the test client
# For server wallet, we'd need a separate private key, but for testing we use test client key
echo "Server wallet funding uses same faucet (test client wallet)"
- name: Start Test Server
run: |
chmod +x scripts/start-test-server.sh
./scripts/start-test-server.sh
- name: Build Test Client
run: |
go build -o test-endpoints ./testClient
- name: Run Endpoint Tests
env:
RPC_URL: ${{ env.RPC_URL }}
FUNDING_FAUCET_URL: ${{ secrets.FUNDING_FAUCET_URL }}
run: |
./test-endpoints
- name: Cleanup
if: always()
run: |
if [ -f /tmp/test-server.pid ]; then
kill $(cat /tmp/test-server.pid) 2>/dev/null || true
fi
docker-compose -f docker-compose.test.yml down -v

4
.gitignore vendored
View file

@ -1,3 +1,5 @@
.env
cdpScripts/cdp
bin/
bin/
test-endpoints
test-endpoints.exe

View file

@ -55,7 +55,7 @@ func main() {
if err != nil {
log.Fatal("Invalid OLLAMA_HOST URL", "error", err)
}
ollamaClient := ollama.New(ctx, uri, http.DefaultClient)
ollamaClient := ollama.New(ctx, uri, http.DefaultClient, cfg.EmbeddingModel, cfg.StyleModel)
apClient, err := activitypub.NewAPClient(ctx, cfg.ActivityPubHost, cfg.ActivityPubToken)
if err != nil {

5
config/models.env Normal file
View file

@ -0,0 +1,5 @@
# Ollama model configuration
# Single source of truth for model names used by the application and CI
EMBEDDING_MODEL=qwen3-embedding:0.6b
STYLE_MODEL=qwen3:4b-instruct-2507-q4_K_M

26
docker-compose.test.yml Normal file
View file

@ -0,0 +1,26 @@
version: '3.8'
services:
postgres:
image: pgvector/pgvector:pg16
environment:
POSTGRES_DB: mcpserver
POSTGRES_USER: postgres
POSTGRES_PASSWORD: postgres
ports:
- "5432:5432"
healthcheck:
test: ["CMD-SHELL", "pg_isready -U postgres -d mcpserver"]
interval: 5s
timeout: 5s
retries: 10
ollama:
image: ollama/ollama:latest
ports:
- "11434:11434"
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:11434/api/tags"]
interval: 10s
timeout: 5s
retries: 10

2
go.mod
View file

@ -23,7 +23,7 @@ require (
github.com/deckarep/golang-set/v2 v2.8.0 // indirect
github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.0 // indirect
github.com/ethereum/c-kzg-4844/v2 v2.1.6 // indirect
github.com/ethereum/go-ethereum v1.17.2 // indirect
github.com/ethereum/go-ethereum v1.17.2
github.com/holiman/uint256 v1.3.2 // indirect
github.com/klauspost/compress v1.18.1 // indirect
github.com/supranational/blst v0.3.16 // indirect

View file

@ -3,10 +3,11 @@ package activitypub
import "time"
type Post struct {
Author string `json:"author"`
Content string `json:"content"`
Text string `json:"text"`
CreatedAt time.Time `json:"created_at"`
ID string `json:"id"`
Author string `json:"author"`
Content string `json:"content"`
Text string `json:"text"`
CreatedAt time.Time `json:"created_at"`
FavouritesCount int `json:"favourites_count"`
ReblogsCount int `json:"reblogs_count"`
RepliesCount int `json:"replies_count"`
@ -14,13 +15,14 @@ type Post struct {
func (i *internalPost) toPost() *Post {
return &Post{
Author: i.Account.Acct,
Content: i.Content,
Text: i.Text,
CreatedAt: i.CreatedAt,
FavouritesCount: i.FavouritesCount,
ReblogsCount: i.ReblogsCount,
RepliesCount: i.RepliesCount,
ID: i.ID,
Author: i.Account.Acct,
Content: i.Content,
Text: i.Text,
CreatedAt: i.CreatedAt,
FavouritesCount: i.FavouritesCount,
ReblogsCount: i.ReblogsCount,
RepliesCount: i.RepliesCount,
}
}

View file

@ -0,0 +1,37 @@
package clustering
import "math"
func normalize(v []float64) []float64 {
out := make([]float64, len(v))
var norm float64
for _, x := range v {
norm += x * x
}
norm = math.Sqrt(norm)
if norm == 0 {
return out
}
for i, x := range v {
out[i] = x / norm
}
return out
}
func cosineSimilarity(a, b []float64) float64 {
if len(a) != len(b) {
return -1
}
var dot float64
for i := range a {
dot += a[i] * b[i]
}
return dot
}

View file

@ -0,0 +1,228 @@
package clustering
import (
"errors"
"math"
"math/rand"
)
func initializeKMeansPlusPlus(
vectors [][]float64,
k int,
rng *rand.Rand,
) [][]float64 {
centroids := make([][]float64, 0, k)
first := rng.Intn(len(vectors))
centroids = append(
centroids,
append([]float64(nil), vectors[first]...),
)
distances := make([]float64, len(vectors))
for len(centroids) < k {
var total float64
for i, vector := range vectors {
bestSimilarity := -math.MaxFloat64
for _, centroid := range centroids {
similarity := cosineSimilarity(vector, centroid)
if similarity > bestSimilarity {
bestSimilarity = similarity
}
}
// cosine distance
distance := 1 - bestSimilarity
if distance < 0 {
distance = 0
}
// k-means++ uses squared distance.
distances[i] = distance * distance
total += distances[i]
}
if total == 0 {
break
}
target := rng.Float64() * total
var cumulative float64
selected := len(vectors) - 1
for i, distance := range distances {
cumulative += distance
if cumulative >= target {
selected = i
break
}
}
centroids = append(
centroids,
append([]float64(nil), vectors[selected]...),
)
}
// This can theoretically happen if all vectors are identical.
for len(centroids) < k {
index := rng.Intn(len(vectors))
centroids = append(
centroids,
append([]float64(nil), vectors[index]...),
)
}
return centroids
}
func KMeans(items []Item, k int, maxIterations int, seed int64) ([]Result, error) {
if len(items) == 0 {
return nil, errors.New("no items")
}
if k <= 0 {
return nil, errors.New("k must be positive")
}
if k > len(items) {
k = len(items)
}
if maxIterations <= 0 {
maxIterations = 50
}
dimension := len(items[0].Embedding)
if dimension == 0 {
return nil, errors.New("empty embedding")
}
// Normalize embeddings once.
normalized := make([][]float64, len(items))
for i, item := range items {
if len(item.Embedding) != dimension {
return nil, errors.New("embedding dimensions do not match")
}
normalized[i] = normalize(item.Embedding)
}
rng := rand.New(rand.NewSource(seed))
centroids := initializeKMeansPlusPlus(normalized, k, rng)
assignments := make([]int, len(items))
for iteration := 0; iteration < maxIterations; iteration++ {
changed := false
// Assignment step.
for i, vector := range normalized {
bestCluster := 0
bestSimilarity := -math.MaxFloat64
for cluster, centroid := range centroids {
similarity := cosineSimilarity(vector, centroid)
if similarity > bestSimilarity {
bestSimilarity = similarity
bestCluster = cluster
}
}
if iteration == 0 || assignments[i] != bestCluster {
assignments[i] = bestCluster
changed = true
}
}
// Stop if nothing changed.
if !changed && iteration > 0 {
break
}
// Update step.
newCentroids := make([][]float64, k)
counts := make([]int, k)
for cluster := 0; cluster < k; cluster++ {
newCentroids[cluster] = make([]float64, dimension)
}
for i, vector := range normalized {
cluster := assignments[i]
counts[cluster]++
for d, value := range vector {
newCentroids[cluster][d] += value
}
}
for cluster := 0; cluster < k; cluster++ {
if counts[cluster] == 0 {
// Empty cluster: reinitialize with a random item.
index := rng.Intn(len(normalized))
newCentroids[cluster] = append(
[]float64(nil),
normalized[index]...,
)
continue
}
newCentroids[cluster] = normalize(newCentroids[cluster])
}
centroids = newCentroids
}
results := make([]Result, k)
for cluster := 0; cluster < k; cluster++ {
results[cluster] = Result{
Cluster: cluster,
Centroid: centroids[cluster],
}
}
for i, item := range items {
cluster := assignments[i]
item.Embedding = normalized[i]
results[cluster].Items = append(
results[cluster].Items,
item,
)
}
// Find the item closest to each centroid.
for cluster := range results {
if len(results[cluster].Items) == 0 {
continue
}
bestSimilarity := -math.MaxFloat64
for _, item := range results[cluster].Items {
similarity := cosineSimilarity(
item.Embedding,
results[cluster].Centroid,
)
if similarity > bestSimilarity {
bestSimilarity = similarity
results[cluster].Representative = item
}
}
}
return results, nil
}

View file

@ -0,0 +1,273 @@
package clustering
import (
"math"
"math/rand"
"testing"
)
func TestKMeans_Basic(t *testing.T) {
items := []Item{
{ID: "1", Text: "a", Embedding: []float64{1.0, 0.0}},
{ID: "2", Text: "b", Embedding: []float64{0.9, 0.1}},
{ID: "3", Text: "c", Embedding: []float64{-1.0, 0.0}},
{ID: "4", Text: "d", Embedding: []float64{-0.9, -0.1}},
}
results, err := KMeans(items, 2, 50, 42)
if err != nil {
t.Fatalf("KMeans failed: %v", err)
}
if len(results) != 2 {
t.Fatalf("expected 2 clusters, got %d", len(results))
}
// Check that items are assigned to clusters
totalItems := 0
for _, r := range results {
totalItems += len(r.Items)
}
if totalItems != len(items) {
t.Fatalf("expected %d items across clusters, got %d", len(items), totalItems)
}
// Verify centroids are normalized
for _, r := range results {
norm := 0.0
for _, v := range r.Centroid {
norm += v * v
}
norm = math.Sqrt(norm)
if math.Abs(norm-1.0) > 1e-6 {
t.Errorf("centroid not normalized: norm=%f", norm)
}
}
}
func TestKMeans_Empty(t *testing.T) {
_, err := KMeans([]Item{}, 2, 50, 42)
if err == nil {
t.Fatal("expected error for empty items")
}
}
func TestKMeans_KGreaterThanItems(t *testing.T) {
items := []Item{
{ID: "1", Text: "a", Embedding: []float64{1.0, 0.0}},
{ID: "2", Text: "b", Embedding: []float64{0.0, 1.0}},
}
results, err := KMeans(items, 5, 50, 42)
if err != nil {
t.Fatalf("KMeans failed: %v", err)
}
// k should be capped to len(items)
if len(results) != 2 {
t.Fatalf("expected 2 clusters (capped), got %d", len(results))
}
}
func TestKMeans_IdenticalVectors(t *testing.T) {
items := []Item{
{ID: "1", Text: "a", Embedding: []float64{1.0, 0.0}},
{ID: "2", Text: "b", Embedding: []float64{1.0, 0.0}},
{ID: "3", Text: "c", Embedding: []float64{1.0, 0.0}},
}
results, err := KMeans(items, 2, 50, 42)
if err != nil {
t.Fatalf("KMeans failed: %v", err)
}
if len(results) != 2 {
t.Fatalf("expected 2 clusters, got %d", len(results))
}
}
func TestKMeans_SingleItem(t *testing.T) {
items := []Item{
{ID: "1", Text: "a", Embedding: []float64{1.0, 0.0}},
}
results, err := KMeans(items, 1, 50, 42)
if err != nil {
t.Fatalf("KMeans failed: %v", err)
}
if len(results) != 1 {
t.Fatalf("expected 1 cluster, got %d", len(results))
}
if len(results[0].Items) != 1 {
t.Fatalf("expected 1 item in cluster, got %d", len(results[0].Items))
}
}
func TestKMeans_DeterministicWithSeed(t *testing.T) {
items := []Item{
{ID: "1", Text: "a", Embedding: []float64{1.0, 0.0}},
{ID: "2", Text: "b", Embedding: []float64{0.9, 0.1}},
{ID: "3", Text: "c", Embedding: []float64{-1.0, 0.0}},
{ID: "4", Text: "d", Embedding: []float64{-0.9, -0.1}},
}
results1, _ := KMeans(items, 2, 50, 42)
results2, _ := KMeans(items, 2, 50, 42)
// Same seed should produce same assignments
for i := range results1 {
if len(results1[i].Items) != len(results2[i].Items) {
t.Fatalf("cluster %d size differs: %d vs %d", i, len(results1[i].Items), len(results2[i].Items))
}
}
}
func TestKMeans_DifferentSeedsMayDiffer(t *testing.T) {
items := []Item{
{ID: "1", Text: "a", Embedding: []float64{1.0, 0.0}},
{ID: "2", Text: "b", Embedding: []float64{0.9, 0.1}},
{ID: "3", Text: "c", Embedding: []float64{-1.0, 0.0}},
{ID: "4", Text: "d", Embedding: []float64{-0.9, -0.1}},
}
results1, _ := KMeans(items, 2, 50, 42)
results2, _ := KMeans(items, 2, 50, 123)
// Different seeds may produce different cluster sizes (though often same for well-separated data)
// Just verify both are valid
for _, r := range results1 {
if len(r.Items) == 0 {
t.Fatal("seed 42 produced empty cluster")
}
}
for _, r := range results2 {
if len(r.Items) == 0 {
t.Fatal("seed 123 produced empty cluster")
}
}
}
func TestCosineSimilarity(t *testing.T) {
tests := []struct {
name string
a, b []float64
expected float64
}{
{"identical", []float64{1, 0}, []float64{1, 0}, 1.0},
{"orthogonal", []float64{1, 0}, []float64{0, 1}, 0.0},
{"opposite", []float64{1, 0}, []float64{-1, 0}, -1.0},
{"scaled", []float64{2, 0}, []float64{1, 0}, 2.0}, // not normalized
{"mismatched len", []float64{1, 0}, []float64{1, 0, 0}, -1.0},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
got := cosineSimilarity(tc.a, tc.b)
if math.Abs(got-tc.expected) > 1e-6 {
t.Errorf("cosineSimilarity(%v, %v) = %f, want %f", tc.a, tc.b, got, tc.expected)
}
})
}
}
func TestNormalize(t *testing.T) {
v := []float64{3.0, 4.0}
normalized := normalize(v)
// Check length is 1
norm := 0.0
for _, x := range normalized {
norm += x * x
}
norm = math.Sqrt(norm)
if math.Abs(norm-1.0) > 1e-6 {
t.Errorf("normalized vector has norm %f, want 1.0", norm)
}
// Check zero vector handling
zero := normalize([]float64{0.0, 0.0})
if zero[0] != 0.0 || zero[1] != 0.0 {
t.Errorf("zero vector normalization failed: %v", zero)
}
}
func TestInitializeKMeansPlusPlus(t *testing.T) {
vectors := [][]float64{
{1.0, 0.0},
{0.9, 0.1},
{-1.0, 0.0},
{-0.9, -0.1},
}
rng := rand.New(rand.NewSource(42))
centroids := initializeKMeansPlusPlus(vectors, 2, rng)
if len(centroids) != 2 {
t.Fatalf("expected 2 centroids, got %d", len(centroids))
}
// Centroids should be from the input vectors (or averages)
found1, found2 := false, false
for _, c := range centroids {
for _, v := range vectors {
if equalVectors(c, v) {
if !found1 {
found1 = true
} else {
found2 = true
}
}
}
}
if !found1 || !found2 {
t.Errorf("centroids not from input vectors: %v", centroids)
}
}
func equalVectors(a, b []float64) bool {
if len(a) != len(b) {
return false
}
for i := range a {
if math.Abs(a[i]-b[i]) > 1e-6 {
return false
}
}
return true
}
func TestKMeans_HighDimensional(t *testing.T) {
// Test with 1024-dim vectors (like qwen3-embedding)
dim := 1024
items := make([]Item, 10)
for i := range items {
emb := make([]float64, dim)
if i < 5 {
emb[0] = 1.0 // cluster 1
} else {
emb[dim-1] = 1.0 // cluster 2
}
items[i] = Item{
ID: string(rune('a' + i)),
Text: "test",
Embedding: emb,
}
}
results, err := KMeans(items, 2, 50, 42)
if err != nil {
t.Fatalf("KMeans failed: %v", err)
}
if len(results) != 2 {
t.Fatalf("expected 2 clusters, got %d", len(results))
}
// Verify each cluster has ~5 items
for _, r := range results {
if len(r.Items) < 3 || len(r.Items) > 7 {
t.Errorf("cluster size %d unexpected", len(r.Items))
}
}
}

View file

@ -0,0 +1,21 @@
package clustering
import "time"
type Item struct {
ID string
Text string
Embedding []float64
Timestamp time.Time
Likes int64
Reposts int64
Replies int64
Quotes int64
}
type Result struct {
Cluster int
Items []Item
Centroid []float64
Representative Item
}

View file

@ -2,8 +2,10 @@ package config
import (
"os"
"path/filepath"
"strconv"
"strings"
"time"
"github.com/joho/godotenv"
log "github.com/nacorid/logger"
@ -38,9 +40,32 @@ type Config struct {
ActiveNetworks string
MatrixWebhookURL string
MatrixServerID string
EmbeddingModel string
StyleModel string
AnalyticsTTLEngagement time.Duration
AnalyticsTTLPatterns time.Duration
AnalyticsTTLTopics time.Duration
AnalyticsTTLSentiment time.Duration
AnalyticsTTLDrivers time.Duration
}
func Load() *Config {
// Load config/models.env for model defaults (single source of truth)
// Find project root by looking for go.mod
root := findProjectRoot()
modelEnvPaths := []string{
filepath.Join(root, "config", "models.env"),
filepath.Join(root, "..", "config", "models.env"), // fallback
}
for _, p := range modelEnvPaths {
if err := godotenv.Load(p); err == nil {
log.Debug("Loaded model config", "path", p)
break
} else {
log.Debug("Model config not found", "path", p, "error", err)
}
}
// Also load .env for local development overrides
if err := godotenv.Load(); err != nil {
log.Debug("No .env file found, reading from environment")
}
@ -66,7 +91,7 @@ func Load() *Config {
PrewarmFeedURI: getEnv("PREWARM_FEED_URI", ""),
EnrichCrossPlatform: getEnvBool("ENRICH_CROSS_PLATFORM", "true"),
OllamaHost: getEnv("OLLAMA_HOST", "http://localhost:11434"),
MigrateOnStart: getEnvBool("MIGRATE_ON_START", "false"),
MigrateOnStart: getEnvBool("MIGRATE_ON_START", "true"),
MigrationsDir: getEnv("MIGRATIONS_DIR", "sql/schema"),
ReadyzPort: getEnv("READYZ_PORT", "9090"),
MetricsPort: getEnv("METRICS_PORT", "9091"),
@ -78,6 +103,13 @@ func Load() *Config {
ActiveNetworks: getEnv("ACTIVE_NETWORKS", "base"),
MatrixWebhookURL: getEnv("MATRIX_WEBHOOK_URL", ""),
MatrixServerID: getEnv("MATRIX_SERVER_ID", ""),
EmbeddingModel: getEnv("EMBEDDING_MODEL", ""),
StyleModel: getEnv("STYLE_MODEL", ""),
AnalyticsTTLEngagement: getEnvDuration("ANALYTICS_TTL_ENGAGEMENT", "1h"),
AnalyticsTTLPatterns: getEnvDuration("ANALYTICS_TTL_PATTERNS", "6h"),
AnalyticsTTLTopics: getEnvDuration("ANALYTICS_TTL_TOPICS", "24h"),
AnalyticsTTLSentiment: getEnvDuration("ANALYTICS_TTL_SENTIMENT", "6h"),
AnalyticsTTLDrivers: getEnvDuration("ANALYTICS_TTL_DRIVERS", "1h"),
}
// Default MatrixServerID to system hostname if not explicitly set
@ -111,12 +143,40 @@ func getEnvBool(key, fallback string) bool {
}
}
func getEnvDuration(key, fallback string) time.Duration {
v := getEnv(key, fallback)
d, err := time.ParseDuration(v)
if err != nil {
d, _ = time.ParseDuration(fallback)
}
return d
}
func splitComma(s string) []string {
parts := make([]string, 0)
for _, p := range strings.Split(s, ",") {
for p := range strings.SplitSeq(s, ",") {
if trimmed := strings.TrimSpace(p); trimmed != "" {
parts = append(parts, trimmed)
}
}
return parts
}
func findProjectRoot() string {
// Start from current working directory and walk up looking for go.mod
dir, err := os.Getwd()
if err != nil {
return "."
}
for {
if _, err := os.Stat(filepath.Join(dir, "go.mod")); err == nil {
return dir
}
parent := filepath.Dir(dir)
if parent == dir {
break // reached filesystem root
}
dir = parent
}
return "."
}

View file

@ -6,39 +6,64 @@ import (
"net/url"
"strings"
"github.com/nacorid/naco-api/internal/clustering"
"github.com/ollama/ollama/api"
)
type OllamaClient struct {
client api.Client
client api.Client
embeddingModel string
styleModel string
}
func New(ctx context.Context, host *url.URL, http *http.Client) *OllamaClient {
func New(ctx context.Context, host *url.URL, http *http.Client, em, sm string) *OllamaClient {
client := api.NewClient(host, http)
return &OllamaClient{client: *client}
return &OllamaClient{client: *client, embeddingModel: em, styleModel: sm}
}
func (o *OllamaClient) EmbedTexts(ctx context.Context, posts []string) ([][]float64, error) {
vectors := make([][]float64, 0, len(posts))
for _, t := range posts {
func (o *OllamaClient) EmbedTexts(ctx context.Context, posts []clustering.Item) ([][]float64, error) {
if len(posts) == 0 {
return nil, nil
}
embeddings := make([][]float64, 0, len(posts))
for _, post := range posts {
resp, err := o.client.Embeddings(ctx, &api.EmbeddingRequest{
Model: "nomic-embed-text",
Prompt: t,
Model: o.embeddingModel,
Prompt: post.Text,
})
if err != nil {
return nil, err
}
vectors = append(vectors, resp.Embedding)
embeddings = append(embeddings, resp.Embedding)
}
return vectors, nil
return embeddings, nil
}
func (o *OllamaClient) SummarizeStyle(ctx context.Context, posts []string) (string, error) {
prompt := "Summarize writing style, recurring themes, tone, personality traits of this user:\n\n" +
strings.Join(posts, "\n") + "\n\nStyle summary:"
func (o *OllamaClient) SummarizeStyle(ctx context.Context, posts []clustering.Item) (string, error) {
if len(posts) == 0 {
return "", nil
}
var promptParts []string
for _, post := range posts {
promptParts = append(promptParts, post.Text)
}
prompt := "Summarize the user's writing style. Analyze:\n" +
"- tone and emotional register\n" +
"- recurring themes and interests\n" +
"- vocabulary and sentence structure\n" +
"- communication habits\n" +
"- personality traits expressed through writing\n\n" +
"Base your analysis only on the supplied posts. Do not infer sensitive personal attributes.\n\n" +
"Posts:\n\n" +
strings.Join(promptParts, "\n---\n") +
"\n\nStyle summary:"
var fullResponse strings.Builder
err := o.client.Generate(ctx, &api.GenerateRequest{
Model: "mistral",
Model: o.styleModel,
Prompt: prompt,
}, func(resp api.GenerateResponse) error {
fullResponse.WriteString(resp.Response)

View file

@ -0,0 +1,260 @@
package ollama
import (
"context"
"net/http"
"net/http/httptest"
"net/url"
"testing"
"github.com/nacorid/naco-api/internal/clustering"
)
func TestEmbedTexts(t *testing.T) {
mux := http.NewServeMux()
mux.HandleFunc("/api/embeddings", func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
w.Write([]byte(`{"embedding":[0.1,0.2,0.3]}`))
})
server := httptest.NewServer(mux)
defer server.Close()
host, _ := url.Parse(server.URL)
client := New(context.Background(), host, http.DefaultClient, "test-embed", "test-style")
posts := []clustering.Item{
{ID: "1", Text: "hello", Embedding: nil},
{ID: "2", Text: "world", Embedding: nil},
}
embeddings, err := client.EmbedTexts(context.Background(), posts)
if err != nil {
t.Fatalf("EmbedTexts failed: %v", err)
}
if len(embeddings) != 2 {
t.Fatalf("expected 2 embeddings, got %d", len(embeddings))
}
for i, emb := range embeddings {
if len(emb) != 3 {
t.Errorf("embedding %d has wrong dim: %d", i, len(emb))
}
if emb[0] != 0.1 || emb[1] != 0.2 || emb[2] != 0.3 {
t.Errorf("embedding %d wrong values: %v", i, emb)
}
}
}
func TestEmbedTexts_Error(t *testing.T) {
mux := http.NewServeMux()
mux.HandleFunc("/api/embeddings", func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusInternalServerError)
w.Write([]byte("server error"))
})
server := httptest.NewServer(mux)
defer server.Close()
host, _ := url.Parse(server.URL)
client := New(context.Background(), host, http.DefaultClient, "test-embed", "test-style")
posts := []clustering.Item{{ID: "1", Text: "hello", Embedding: nil}}
_, err := client.EmbedTexts(context.Background(), posts)
if err == nil {
t.Fatal("expected error from failing server")
}
}
func TestSummarizeStyle(t *testing.T) {
mux := http.NewServeMux()
mux.HandleFunc("/api/generate", func(w http.ResponseWriter, r *http.Request) {
flusher, ok := w.(http.Flusher)
if !ok {
t.Fatal("response writer doesn't support flushing")
}
w.Header().Set("Content-Type", "application/x-ndjson")
// Stream response as newline-delimited JSON
w.Write([]byte(`{"response":"Style summary: "}` + "\n"))
if flusher != nil {
flusher.Flush()
}
w.Write([]byte(`{"response":"concise and clear."}` + "\n"))
})
server := httptest.NewServer(mux)
defer server.Close()
host, _ := url.Parse(server.URL)
client := New(context.Background(), host, http.DefaultClient, "test-embed", "test-style")
posts := []clustering.Item{
{ID: "1", Text: "First post", Embedding: nil},
{ID: "2", Text: "Second post", Embedding: nil},
}
summary, err := client.SummarizeStyle(context.Background(), posts)
if err != nil {
t.Fatalf("SummarizeStyle failed: %v", err)
}
if summary != "Style summary: concise and clear." {
t.Errorf("unexpected summary: %q", summary)
}
}
func TestSummarizeStyle_Error(t *testing.T) {
mux := http.NewServeMux()
mux.HandleFunc("/api/generate", func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusInternalServerError)
w.Write([]byte("server error"))
})
server := httptest.NewServer(mux)
defer server.Close()
host, _ := url.Parse(server.URL)
client := New(context.Background(), host, http.DefaultClient, "test-embed", "test-style")
posts := []clustering.Item{{ID: "1", Text: "hello", Embedding: nil}}
_, err := client.SummarizeStyle(context.Background(), posts)
if err == nil {
t.Fatal("expected error from failing server")
}
}
func TestSummarizeStyle_PromptStructure(t *testing.T) {
var receivedPrompt string
mux := http.NewServeMux()
mux.HandleFunc("/api/generate", func(w http.ResponseWriter, r *http.Request) {
// Capture the prompt from request body
receivedPrompt = "captured"
w.Header().Set("Content-Type", "application/json")
w.Write([]byte(`{"response":"ok"}`))
})
server := httptest.NewServer(mux)
defer server.Close()
host, _ := url.Parse(server.URL)
client := New(context.Background(), host, http.DefaultClient, "test-embed", "test-style")
posts := []clustering.Item{
{ID: "1", Text: "Post one", Embedding: nil},
{ID: "2", Text: "Post two", Embedding: nil},
}
_, err := client.SummarizeStyle(context.Background(), posts)
if err != nil {
t.Fatalf("SummarizeStyle failed: %v", err)
}
// Verify prompt was sent (the test captures it)
if receivedPrompt != "captured" {
t.Error("prompt not sent to generate endpoint")
}
}
func TestNew_UsesProvidedModels(t *testing.T) {
host, _ := url.Parse("http://localhost:11434")
client := New(context.Background(), host, http.DefaultClient, "custom-embed", "custom-style")
if client.embeddingModel != "custom-embed" {
t.Errorf("embeddingModel = %q, want %q", client.embeddingModel, "custom-embed")
}
if client.styleModel != "custom-style" {
t.Errorf("styleModel = %q, want %q", client.styleModel, "custom-style")
}
}
// Test with actual api.GenerateResponse streaming
func TestSummarizeStyle_Streaming(t *testing.T) {
callCount := 0
mux := http.NewServeMux()
mux.HandleFunc("/api/generate", func(w http.ResponseWriter, r *http.Request) {
callCount++
flusher, ok := w.(http.Flusher)
if !ok {
t.Fatal("response writer doesn't support flushing")
}
w.Header().Set("Content-Type", "application/x-ndjson")
// Simulate streaming response
w.Write([]byte(`{"response":"part1"}` + "\n"))
if flusher != nil {
flusher.Flush()
}
w.Write([]byte(`{"response":"part2"}` + "\n"))
if flusher != nil {
flusher.Flush()
}
w.Write([]byte(`{"response":"part3","done":true}` + "\n"))
})
server := httptest.NewServer(mux)
defer server.Close()
host, _ := url.Parse(server.URL)
client := New(context.Background(), host, http.DefaultClient, "test-embed", "test-style")
posts := []clustering.Item{{ID: "1", Text: "test", Embedding: nil}}
summary, err := client.SummarizeStyle(context.Background(), posts)
if err != nil {
t.Fatalf("SummarizeStyle failed: %v", err)
}
if summary != "part1part2part3" {
t.Errorf("streaming summary = %q, want %q", summary, "part1part2part3")
}
if callCount != 1 {
t.Errorf("generate endpoint called %d times, want 1", callCount)
}
}
func TestEmbedTexts_EmptyInput(t *testing.T) {
mux := http.NewServeMux()
mux.HandleFunc("/api/embeddings", func(w http.ResponseWriter, r *http.Request) {
t.Fatal("embeddings endpoint should not be called for empty input")
})
server := httptest.NewServer(mux)
defer server.Close()
host, _ := url.Parse(server.URL)
client := New(context.Background(), host, http.DefaultClient, "test-embed", "test-style")
embeddings, err := client.EmbedTexts(context.Background(), []clustering.Item{})
if err != nil {
t.Fatalf("EmbedTexts with empty input failed: %v", err)
}
if len(embeddings) != 0 {
t.Errorf("expected empty embeddings, got %d", len(embeddings))
}
}
func TestSummarizeStyle_EmptyInput(t *testing.T) {
mux := http.NewServeMux()
mux.HandleFunc("/api/generate", func(w http.ResponseWriter, r *http.Request) {
t.Fatal("generate endpoint should not be called for empty input")
})
server := httptest.NewServer(mux)
defer server.Close()
host, _ := url.Parse(server.URL)
client := New(context.Background(), host, http.DefaultClient, "test-embed", "test-style")
summary, err := client.SummarizeStyle(context.Background(), []clustering.Item{})
if err != nil {
t.Fatalf("SummarizeStyle with empty input failed: %v", err)
}
if summary != "" {
t.Errorf("expected empty summary, got %q", summary)
}
}

View file

@ -0,0 +1,883 @@
package server
import (
"context"
"crypto/sha256"
"database/sql"
"encoding/hex"
"encoding/json"
"fmt"
"math"
"sort"
"strings"
"time"
"github.com/bluesky-social/indigo/api/bsky"
"github.com/modelcontextprotocol/go-sdk/mcp"
log "github.com/nacorid/logger"
"github.com/nacorid/naco-api/internal/clustering"
"github.com/nacorid/naco-api/internal/store"
"github.com/nacorid/naco-api/internal/utils"
)
const (
defaultAnalyticsLimit = 50
maxAnalyticsLimit = 200
defaultAnalyticsDays = 30
maxTopicCount = 10
minTopicCount = 3
sentimentBucketDuration = 24 * time.Hour
)
// ---- Analytics request/response helpers ----
func parseAnalyticsTimeRange(startStr, endStr string) (time.Time, time.Time, error) {
now := time.Now().UTC()
end := now
start := now.AddDate(0, 0, -defaultAnalyticsDays)
if endStr != "" {
if t, err := time.Parse(time.RFC3339, endStr); err == nil {
end = t
}
}
if startStr != "" {
if t, err := time.Parse(time.RFC3339, startStr); err == nil {
start = t
}
}
if start.After(end) {
start, end = end, start
}
return start, end, nil
}
func hashTimeRange(start, end time.Time, limit int) string {
data := fmt.Sprintf("%s|%s|%d", start.Format(time.RFC3339), end.Format(time.RFC3339), limit)
h := sha256.Sum256([]byte(data))
return hex.EncodeToString(h[:])
}
func (s *Server) getAnalyticsCache(ctx context.Context, author, platform, toolName, timeRangeHash string) (json.RawMessage, bool) {
result, err := s.db.GetAnalyticsCache(ctx, store.GetAnalyticsCacheParams{
UserID: author,
Platform: platform,
ToolName: toolName,
TimeRangeHash: timeRangeHash,
})
if err != nil {
return nil, false
}
return result, true
}
func (s *Server) setAnalyticsCache(ctx context.Context, author, platform, toolName, timeRangeHash string, result any, ttl time.Duration) error {
data, err := json.Marshal(result)
if err != nil {
return err
}
expiresAt := time.Now().Add(ttl)
return s.db.UpsertAnalyticsCache(ctx, store.UpsertAnalyticsCacheParams{
UserID: author,
Platform: platform,
ToolName: toolName,
TimeRangeHash: timeRangeHash,
Result: data,
ExpiresAt: expiresAt,
})
}
// ---- Main analytics entry points ----
func (s *Server) getEngagementMetrics(ctx context.Context, request *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
var args AnalyticsRequest
if err := json.Unmarshal(request.Params.Arguments, &args); err != nil {
return nil, fmt.Errorf("invalid arguments: %w", err)
}
return s.computeAnalytics(ctx, args, "engagementMetrics", s.computeEngagementMetrics)
}
func (s *Server) getPostingPatterns(ctx context.Context, request *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
var args AnalyticsRequest
if err := json.Unmarshal(request.Params.Arguments, &args); err != nil {
return nil, fmt.Errorf("invalid arguments: %w", err)
}
return s.computeAnalytics(ctx, args, "postingPatterns", s.computePostingPatterns)
}
func (s *Server) getTopicModeling(ctx context.Context, request *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
var args AnalyticsRequest
if err := json.Unmarshal(request.Params.Arguments, &args); err != nil {
return nil, fmt.Errorf("invalid arguments: %w", err)
}
return s.computeAnalytics(ctx, args, "topicModeling", s.computeTopicModeling)
}
func (s *Server) getSentimentTrends(ctx context.Context, request *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
var args AnalyticsRequest
if err := json.Unmarshal(request.Params.Arguments, &args); err != nil {
return nil, fmt.Errorf("invalid arguments: %w", err)
}
return s.computeAnalytics(ctx, args, "sentimentTrends", s.computeSentimentTrends)
}
func (s *Server) getEngagementDrivers(ctx context.Context, request *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
var args AnalyticsRequest
if err := json.Unmarshal(request.Params.Arguments, &args); err != nil {
return nil, fmt.Errorf("invalid arguments: %w", err)
}
return s.computeAnalytics(ctx, args, "engagementDrivers", s.computeEngagementDrivers)
}
type analyticsComputer func(ctx context.Context, args AnalyticsRequest, posts []clustering.Item) (any, error)
func (s *Server) computeAnalytics(ctx context.Context, args AnalyticsRequest, toolName string, compute analyticsComputer) (*mcp.CallToolResult, error) {
// Normalize args
limit := args.Limit
if limit <= 0 {
limit = defaultAnalyticsLimit
}
if limit > maxAnalyticsLimit {
limit = maxAnalyticsLimit
}
start, end, err := parseAnalyticsTimeRange(args.StartTime, args.EndTime)
if err != nil {
return nil, fmt.Errorf("invalid time range: %w", err)
}
platforms := []string{}
if args.Platform != "" {
platforms = []string{args.Platform}
} else {
platforms = []string{"Bluesky", "ActivityPub", "Nostr"}
}
timeRangeHash := hashTimeRange(start, end, limit)
// The analytics result cache is keyed by the scope of the computation: a
// single platform when one is requested, or the "all" sentinel when the
// result aggregates across all platforms. This keeps aggregate results from
// being mixed with single-platform results.
cachePlatform := "all"
if args.Platform != "" {
cachePlatform = args.Platform
}
if cached, ok := s.getAnalyticsCache(ctx, args.Author, cachePlatform, toolName, timeRangeHash); ok {
var cachedResult json.RawMessage
json.Unmarshal(cached, &cachedResult)
return s.toToolResult(cachedResult)
}
// Fetch posts from all platforms
var allPosts []clustering.Item
for _, p := range platforms {
plat, ok := platformFromString(p)
if !ok {
continue
}
posts, err := s.fetchAnalyticsPosts(ctx, plat, args.Author, start, end, limit)
if err != nil {
log.WarnWithContext(ctx, "failed to fetch posts for analytics", "platform", p, "error", err)
continue
}
allPosts = append(allPosts, posts...)
}
if len(allPosts) == 0 {
return nil, fmt.Errorf("no posts found for author %s in time range", args.Author)
}
// Compute
result, err := compute(ctx, args, allPosts)
if err != nil {
return nil, err
}
// Cache the result under the matching scope key.
ttl := s.getCacheTTL(toolName)
s.setAnalyticsCache(ctx, args.Author, cachePlatform, toolName, timeRangeHash, result, ttl)
return s.toToolResult(result)
}
func (s *Server) fetchAnalyticsPosts(ctx context.Context, platform platform, author string, start, end time.Time, limit int) ([]clustering.Item, error) {
posts, err := s.fetchPostsInTimeForAnalytics(ctx, platform, author, start, end, limit)
if err != nil {
return nil, err
}
return posts, nil
}
func (s *Server) getCachedAnalyticsPosts(ctx context.Context, author, platform string, start, end time.Time, limit int) ([]clustering.Item, error) {
rows, err := s.db.GetAnalyticsPostsInRange(ctx, store.GetAnalyticsPostsInRangeParams{
UserID: author,
Platform: platform,
PostTimestamp: start,
PostTimestamp_2: end,
Limit: int32(limit),
})
if err != nil {
return nil, err
}
items := make([]clustering.Item, 0, len(rows))
for _, r := range rows {
items = append(items, clustering.Item{
ID: r.PostID,
Text: r.PostText,
Timestamp: r.PostTimestamp,
Likes: r.Likes,
Reposts: r.Reposts,
Replies: r.Replies,
Quotes: r.Quotes,
})
}
return items, nil
}
func (s *Server) persistAnalyticsPosts(ctx context.Context, author, platform string, posts []clustering.Item) error {
for _, p := range posts {
if p.ID == "" {
continue
}
if err := s.db.UpsertAnalyticsPosts(ctx, store.UpsertAnalyticsPostsParams{
UserID: author,
Platform: platform,
PostID: p.ID,
PostText: p.Text,
PostTimestamp: p.Timestamp,
Likes: p.Likes,
Reposts: p.Reposts,
Replies: p.Replies,
Quotes: p.Quotes,
}); err != nil {
return err
}
}
return nil
}
func (s *Server) fetchPostsInTimeForAnalytics(ctx context.Context, platform platform, author string, start, end time.Time, limit int) ([]clustering.Item, error) {
var items []clustering.Item
switch platform {
case Bluesky:
res, err := s.bskyClient.GetPostsInTimeframe(ctx, author, start, end, limit)
if err != nil {
return nil, err
}
for _, p := range res {
if p.Post == nil || p.Post.Record == nil {
continue
}
if fp, ok := p.Post.Record.Val.(*bsky.FeedPost); ok {
item := clustering.Item{
ID: p.Post.Uri,
Text: fp.Text,
}
if ts, err := time.Parse(time.RFC3339, p.Post.IndexedAt); err == nil {
item.Timestamp = ts
}
item.Likes = int64Val(p.Post.LikeCount)
item.Reposts = int64Val(p.Post.RepostCount)
item.Replies = int64Val(p.Post.ReplyCount)
item.Quotes = int64Val(p.Post.QuoteCount)
items = append(items, item)
}
}
case ActivityPub:
res, err := s.apClient.GetPostsInTimeframe(ctx, author, start, end, limit)
if err != nil {
return nil, err
}
for _, p := range res {
if p.Text != "" {
items = append(items, clustering.Item{
ID: p.ID,
Text: p.Text,
Timestamp: p.CreatedAt,
Likes: int64(p.FavouritesCount),
Reposts: int64(p.ReblogsCount),
Replies: int64(p.RepliesCount),
})
}
}
case Nostr:
res, err := s.nostrClient.GetPostsInTimeframe(ctx, author, start, end, limit)
if err != nil {
return nil, err
}
for _, p := range res {
if p.Content != "" {
items = append(items, clustering.Item{
ID: p.ID,
Text: p.Content,
Timestamp: p.CreatedAt,
})
}
}
case Twitter:
// Twitter API calls are expensive (paid). Reuse persisted post data and
// only call the API for time ranges that are not yet covered by the
// cache, using a coverage watermark to guarantee broad coverage.
fetched, err := s.fetchTwitterAnalyticsPosts(ctx, author, start, end, limit)
if err != nil {
return nil, err
}
items = append(items, fetched...)
}
return items, nil
}
// fetchTwitterAnalyticsPosts returns posts in [start, end] for the author,
// fetching from the Twitter API only the sub-ranges not already covered by the
// persistent post cache, and storing any newly fetched posts for future use.
func (s *Server) fetchTwitterAnalyticsPosts(ctx context.Context, author string, start, end time.Time, limit int) ([]clustering.Item, error) {
coverage, err := s.getAnalyticsCoverage(ctx, author, "Twitter")
if err != nil {
// Unknown coverage (missing row) is not an error; fall back to fetching.
if err != sql.ErrNoRows {
return nil, err
}
}
coverageStart, coverageEnd := time.Time{}, time.Time{}
haveCoverage := !coverage.CoveredStart.IsZero() && !coverage.CoveredEnd.IsZero()
if haveCoverage {
coverageStart, coverageEnd = coverage.CoveredStart, coverage.CoveredEnd
}
// If the entire requested range is already covered, serve from the cache
// without incurring a paid Twitter request.
if haveCoverage && !start.Before(coverageStart) && !end.After(coverageEnd) {
return s.getCachedAnalyticsPosts(ctx, author, "Twitter", start, end, limit)
}
// Determine the sub-ranges of [start, end] not yet covered, and fetch each
// from the API. Fetching per uncovered range keeps costs bounded while
// expanding coverage on every call.
ranges := uncoveredRanges(start, end, coverageStart, coverageEnd, haveCoverage)
var fetched []clustering.Item
for _, r := range ranges {
posts, err := s.fetchTwitterFromAPI(ctx, author, r.start, r.end, limit)
if err != nil {
return nil, err
}
fetched = append(fetched, posts...)
}
// Persist newly fetched posts permanently so repeat calls never re-pay.
if len(fetched) > 0 {
if err := s.persistAnalyticsPosts(ctx, author, "Twitter", fetched); err != nil {
log.WarnWithContext(ctx, "failed to persist analytics posts", "platform", "Twitter", "error", err)
}
}
// Extend coverage to encompass the whole requested range.
if err := s.extendAnalyticsCoverage(ctx, author, "Twitter", start, end); err != nil {
log.WarnWithContext(ctx, "failed to extend analytics coverage", "platform", "Twitter", "error", err)
}
// Return the full set: posts from cache plus anything newly fetched, merged
// and de-duplicated by post ID.
return s.mergeAnalyticsPosts(ctx, author, start, end, limit, fetched)
}
type timeRange struct {
start, end time.Time
}
// uncoveredRanges returns the sub-ranges of [start, end] not covered by
// [coveredStart, coveredEnd]. If there is no coverage, the whole range is
// returned.
func uncoveredRanges(start, end, coveredStart, coveredEnd time.Time, have bool) []timeRange {
if !have {
return []timeRange{{start: start, end: end}}
}
var ranges []timeRange
if start.Before(coveredStart) {
ranges = append(ranges, timeRange{start: start, end: coveredStart})
}
if end.After(coveredEnd) {
ranges = append(ranges, timeRange{start: coveredEnd, end: end})
}
return ranges
}
func (s *Server) getAnalyticsCoverage(ctx context.Context, author, platform string) (store.GetAnalyticsCoverageRow, error) {
return s.db.GetAnalyticsCoverage(ctx, store.GetAnalyticsCoverageParams{
UserID: author,
Platform: platform,
})
}
func (s *Server) extendAnalyticsCoverage(ctx context.Context, author, platform string, start, end time.Time) error {
return s.db.UpsertAnalyticsCoverage(ctx, store.UpsertAnalyticsCoverageParams{
UserID: author,
Platform: platform,
CoveredStart: start,
CoveredEnd: end,
})
}
func (s *Server) fetchTwitterFromAPI(ctx context.Context, author string, start, end time.Time, limit int) ([]clustering.Item, error) {
res, err := s.twitterClient.GetPostsInTimeframe(ctx, author, start, end, limit)
if err != nil {
return nil, err
}
var items []clustering.Item
for _, p := range res {
if p.Content == "" {
continue
}
items = append(items, clustering.Item{
ID: p.ID,
Text: p.Content,
Timestamp: p.CreatedAt,
Likes: int64(p.Likes),
Reposts: int64(p.Reposts),
Replies: int64(p.Replies),
Quotes: int64(p.Quotes),
})
}
return items, nil
}
// mergeAnalyticsPosts combines cached posts in [start, end] with newly fetched
// posts, de-duplicating by post ID.
func (s *Server) mergeAnalyticsPosts(ctx context.Context, author string, start, end time.Time, limit int, fetched []clustering.Item) ([]clustering.Item, error) {
merged := make([]clustering.Item, 0, len(fetched))
seen := make(map[string]struct{}, len(fetched))
for _, p := range fetched {
if _, ok := seen[p.ID]; ok {
continue
}
seen[p.ID] = struct{}{}
merged = append(merged, p)
}
cached, err := s.getCachedAnalyticsPosts(ctx, author, "Twitter", start, end, limit)
if err != nil {
return merged, nil
}
for _, p := range cached {
if _, ok := seen[p.ID]; ok {
continue
}
seen[p.ID] = struct{}{}
merged = append(merged, p)
}
return merged, nil
}
func (s *Server) toToolResult(result any) (*mcp.CallToolResult, error) {
data, err := json.Marshal(result)
if err != nil {
return nil, err
}
return &mcp.CallToolResult{
Content: []mcp.Content{
&mcp.TextContent{Text: string(data)},
},
}, nil
}
func (s *Server) getCacheTTL(toolName string) time.Duration {
switch toolName {
case "postingPatterns":
return s.analyticsTTLPatterns
case "topicModeling":
return s.analyticsTTLTopics
case "sentimentTrends":
return s.analyticsTTLSentiment
case "engagementDrivers":
return s.analyticsTTLDrivers
case "engagementMetrics":
fallthrough
default:
return s.analyticsTTLEngagement
}
}
// ---- Computation implementations ----
func (s *Server) computeEngagementMetrics(ctx context.Context, args AnalyticsRequest, posts []clustering.Item) (any, error) {
var metrics []EngagementMetricPoint
for _, post := range posts {
total := post.Likes + post.Reposts + post.Replies + post.Quotes
// Rate is a simple relative measure (interactions per post). Without a
// follower/fan base to normalize against, report total interactions.
metrics = append(metrics, EngagementMetricPoint{
PostID: post.ID,
Timestamp: post.Timestamp.Format(time.RFC3339),
Likes: post.Likes,
Reposts: post.Reposts,
Replies: post.Replies,
Quotes: post.Quotes,
EngagementRate: float64(total),
})
}
return EngagementMetricsResponse{
Author: args.Author,
Metrics: metrics,
}, nil
}
func (s *Server) computePostingPatterns(ctx context.Context, args AnalyticsRequest, posts []clustering.Item) (any, error) {
// Heatmap: 24 hours x 7 days
heatmap := make([][]int, 24)
for i := range heatmap {
heatmap[i] = make([]int, 7)
}
dayCounts := make(map[int]int)
hourCounts := make(map[int]int)
dateSet := make(map[string]bool)
for _, post := range posts {
t := post.Timestamp
if t.IsZero() {
continue
}
hour := t.Hour()
weekday := int(t.Weekday())
heatmap[hour][weekday]++
hourCounts[hour]++
dayCounts[weekday]++
dateSet[t.Format("2006-01-02")] = true
}
// Find peak hours (top 3)
type hourCount struct {
Hour int
Count int
}
var hours []hourCount
for h, c := range hourCounts {
hours = append(hours, hourCount{h, c})
}
sort.Slice(hours, func(i, j int) bool { return hours[i].Count > hours[j].Count })
peakHours := make([]int, 0, 3)
for i := 0; i < len(hours) && i < 3; i++ {
peakHours = append(peakHours, hours[i].Hour)
}
// Find peak days (top 3)
type dayCount struct {
Day int
Count int
}
var days []dayCount
for d, c := range dayCounts {
days = append(days, dayCount{d, c})
}
sort.Slice(days, func(i, j int) bool { return days[i].Count > days[j].Count })
peakDays := make([]int, 0, 3)
for i := 0; i < len(days) && i < 3; i++ {
peakDays = append(peakDays, days[i].Day)
}
totalPosts := len(posts)
postingDays := len(dateSet)
avgPostsPerDay := 0.0
if postingDays > 0 {
avgPostsPerDay = float64(totalPosts) / float64(postingDays)
}
return PostingPatternsResponse{
Author: args.Author,
Heatmap: heatmap,
PeakHours: peakHours,
PeakDays: peakDays,
AvgPostsPerDay: avgPostsPerDay,
TotalPosts: totalPosts,
PostingDays: postingDays,
}, nil
}
func (s *Server) computeTopicModeling(ctx context.Context, args AnalyticsRequest, posts []clustering.Item) (any, error) {
if len(posts) < minTopicCount {
return TopicModelingResponse{Author: args.Author, Topics: []TopicCluster{}}, nil
}
// Use embeddings from ollama for clustering
items := posts
// We need embeddings for clustering - compute them
embeddings, err := s.ollama.EmbedTexts(ctx, items)
if err != nil {
return nil, fmt.Errorf("failed to compute embeddings: %w", err)
}
// Update items with embeddings
for i := range items {
items[i].Embedding = embeddings[i]
}
// Determine optimal number of topics
topicCount := maxTopicCount
if len(items) < maxTopicCount {
topicCount = len(items)
}
if topicCount < minTopicCount {
topicCount = minTopicCount
}
// Run KMeans
results, err := clustering.KMeans(items, topicCount, 50, 42)
if err != nil {
return nil, fmt.Errorf("kmeans failed: %w", err)
}
// Generate labels for each cluster using Ollama
var topics []TopicCluster
for _, cluster := range results {
if len(cluster.Items) == 0 {
continue
}
// Collect sample texts for labeling
var samples []string
for _, item := range cluster.Items {
if len(samples) >= 3 {
break
}
samples = append(samples, item.Text)
}
// Generate label using Ollama
label := s.generateTopicLabel(ctx, samples)
keywords := s.extractKeywords(cluster.Items)
topics = append(topics, TopicCluster{
TopicID: cluster.Cluster,
Label: label,
Keywords: keywords,
PostCount: len(cluster.Items),
Percentage: float64(len(cluster.Items)) / float64(len(items)) * 100,
SamplePosts: samples,
})
}
// Sort by post count descending
sort.Slice(topics, func(i, j int) bool { return topics[i].PostCount > topics[j].PostCount })
return TopicModelingResponse{
Author: args.Author,
Topics: topics,
}, nil
}
func (s *Server) computeSentimentTrends(ctx context.Context, args AnalyticsRequest, posts []clustering.Item) (any, error) {
// Bucket posts by day
bucketMap := make(map[string][]string)
for _, post := range posts {
if post.Timestamp.IsZero() {
continue
}
bucket := post.Timestamp.Truncate(sentimentBucketDuration).Format(time.RFC3339)
bucketMap[bucket] = append(bucketMap[bucket], post.Text)
}
var timeSeries []SentimentPoint
overallPos, overallNeg, overallNeu := 0.0, 0.0, 0.0
emotionCounts := make(map[string]int)
var shifts []SentimentShift
var prevSentiment string
var prevScores [3]float64
// Process buckets in chronological order so sentiment shifts are detected correctly.
buckets := make([]string, 0, len(bucketMap))
for b := range bucketMap {
buckets = append(buckets, b)
}
sort.Strings(buckets)
for _, bucket := range buckets {
texts := bucketMap[bucket]
pos, neg, neu, emotions := s.analyzeSentiment(ctx, texts)
timeSeries = append(timeSeries, SentimentPoint{
Timestamp: bucket,
Positive: pos,
Negative: neg,
Neutral: neu,
PostCount: len(texts),
})
overallPos += pos * float64(len(texts))
overallNeg += neg * float64(len(texts))
overallNeu += neu * float64(len(texts))
for e, c := range emotions {
emotionCounts[e] += c
}
// Detect sentiment shifts
dominant := maxSentiment(pos, neg, neu)
if prevSentiment != "" && dominant != prevSentiment {
mag := math.Abs(pos-prevScores[0]) + math.Abs(neg-prevScores[1]) + math.Abs(neu-prevScores[2])
shifts = append(shifts, SentimentShift{
Timestamp: bucket,
FromSentiment: prevSentiment,
ToSentiment: dominant,
Magnitude: mag,
})
}
prevSentiment = dominant
prevScores = [3]float64{pos, neg, neu}
}
totalPosts := float64(len(posts))
if totalPosts > 0 {
overallPos /= totalPosts
overallNeg /= totalPosts
overallNeu /= totalPosts
}
// Top emotions
var topEmotions []EmotionScore
type emoCount struct {
Emotion string
Count int
}
var emos []emoCount
for e, c := range emotionCounts {
emos = append(emos, emoCount{e, c})
}
sort.Slice(emos, func(i, j int) bool { return emos[i].Count > emos[j].Count })
for i := 0; i < len(emos) && i < 5; i++ {
topEmotions = append(topEmotions, EmotionScore{
Emotion: emos[i].Emotion,
Score: float64(emos[i].Count) / totalPosts,
})
}
return SentimentTrendsResponse{
Author: args.Author,
TimeSeries: timeSeries,
OverallSentiment: SentimentSummary{
Positive: overallPos,
Negative: overallNeg,
Neutral: overallNeu,
},
DominantEmotions: topEmotions,
SentimentShifts: shifts,
}, nil
}
func (s *Server) computeEngagementDrivers(ctx context.Context, args AnalyticsRequest, posts []clustering.Item) (any, error) {
// Topic modeling first
topicResult, err := s.computeTopicModeling(ctx, args, posts)
if err != nil {
return nil, err
}
topicResp := topicResult.(TopicModelingResponse)
var drivers []EngagementDriver
for _, topic := range topicResp.Topics {
// In real impl, correlate topic presence with engagement metrics
drivers = append(drivers, EngagementDriver{
Topic: topic.Label,
AvgEngagement: 0.0,
Correlation: 0.0,
PostCount: topic.PostCount,
SamplePosts: topic.SamplePosts[:min(2, len(topic.SamplePosts))],
})
}
sort.Slice(drivers, func(i, j int) bool {
return math.Abs(drivers[i].Correlation) > math.Abs(drivers[j].Correlation)
})
return EngagementDriversResponse{
Author: args.Author,
Drivers: drivers,
}, nil
}
// ---- Helper functions ----
func (s *Server) generateTopicLabel(ctx context.Context, samples []string) string {
if len(samples) == 0 {
return "General"
}
prompt := fmt.Sprintf(`Given these sample posts, provide a concise 1-3 word topic label:
Posts:
%s
Label only (no explanation):`, strings.Join(samples, "\n---\n"))
resp, err := s.ollama.SummarizeStyle(ctx, []clustering.Item{{Text: prompt}})
if err != nil {
return "General"
}
return strings.TrimSpace(resp)
}
func (s *Server) extractKeywords(items []clustering.Item) []string {
var allText []string
for _, item := range items {
allText = append(allText, item.Text)
}
return utils.TopWords(allText, make(map[string]struct{}), 5)
}
func (s *Server) analyzeSentiment(ctx context.Context, texts []string) (pos, neg, neu float64, emotions map[string]int) {
if len(texts) == 0 {
return 0, 0, 0, nil
}
// Use Ollama for sentiment analysis
prompt := fmt.Sprintf(`Analyze sentiment and emotions in these texts. Return JSON only:
{
"positive": 0.0-1.0,
"negative": 0.0-1.0,
"neutral": 0.0-1.0,
"emotions": {"emotion": count}
}
Texts:
%s`, strings.Join(texts, "\n---\n"))
resp, err := s.ollama.SummarizeStyle(ctx, []clustering.Item{{Text: prompt}})
if err != nil {
return 0.33, 0.33, 0.34, nil
}
var result struct {
Positive float64 `json:"positive"`
Negative float64 `json:"negative"`
Neutral float64 `json:"neutral"`
Emotions map[string]int `json:"emotions"`
}
json.Unmarshal([]byte(resp), &result)
// Normalize
sum := result.Positive + result.Negative + result.Neutral
if sum > 0 {
result.Positive /= sum
result.Negative /= sum
result.Neutral /= sum
}
return result.Positive, result.Negative, result.Neutral, result.Emotions
}
func maxSentiment(pos, neg, neu float64) string {
if pos >= neg && pos >= neu {
return "positive"
}
if neg >= pos && neg >= neu {
return "negative"
}
return "neutral"
}
func int64Val(p *int64) int64 {
if p == nil {
return 0
}
return *p
}

View file

@ -8,6 +8,7 @@ import (
"github.com/bluesky-social/indigo/atproto/syntax"
"github.com/nacorid/naco-api/internal/activitypub"
"github.com/nacorid/naco-api/internal/actor"
"github.com/nacorid/naco-api/internal/clustering"
"github.com/nacorid/naco-api/internal/nostr"
"github.com/nacorid/naco-api/internal/twitter"
)
@ -49,6 +50,6 @@ type twitterBackend interface {
}
type ollamaBackend interface {
EmbedTexts(ctx context.Context, posts []string) ([][]float64, error)
SummarizeStyle(ctx context.Context, posts []string) (string, error)
EmbedTexts(ctx context.Context, posts []clustering.Item) ([][]float64, error)
SummarizeStyle(ctx context.Context, posts []clustering.Item) (string, error)
}

View file

@ -0,0 +1,71 @@
package server
import (
"testing"
"time"
)
func rangeOf(t *testing.T, s, e string) timeRange {
t.Helper()
layout := "2006-01-02 15:04"
start, err := time.Parse(layout, s)
if err != nil {
t.Fatalf("bad start %q: %v", s, err)
}
end, err := time.Parse(layout, e)
if err != nil {
t.Fatalf("bad end %q: %v", e, err)
}
return timeRange{start: start, end: end}
}
func assertRanges(t *testing.T, got []timeRange, want ...timeRange) {
t.Helper()
if len(got) != len(want) {
t.Fatalf("got %d ranges, want %d: %+v", len(got), len(want), got)
}
for i := range want {
if !got[i].start.Equal(want[i].start) || !got[i].end.Equal(want[i].end) {
t.Fatalf("range %d: got [%v, %v], want [%v, %v]",
i, got[i].start, got[i].end, want[i].start, want[i].end)
}
}
}
func TestUncoveredRangesNoCoverage(t *testing.T) {
r := rangeOf(t, "2026-01-01 00:00", "2026-01-31 00:00")
got := uncoveredRanges(r.start, r.end, time.Time{}, time.Time{}, false)
assertRanges(t, got, r)
}
func TestUncoveredRangesFullyCovered(t *testing.T) {
req := rangeOf(t, "2026-01-10 00:00", "2026-01-20 00:00")
cov := rangeOf(t, "2026-01-01 00:00", "2026-01-31 00:00")
got := uncoveredRanges(req.start, req.end, cov.start, cov.end, true)
assertRanges(t, got)
}
func TestUncoveredRangesLeftOnly(t *testing.T) {
req := rangeOf(t, "2026-01-01 00:00", "2026-01-31 00:00")
cov := rangeOf(t, "2026-01-15 00:00", "2026-01-31 00:00")
want := rangeOf(t, "2026-01-01 00:00", "2026-01-15 00:00")
got := uncoveredRanges(req.start, req.end, cov.start, cov.end, true)
assertRanges(t, got, want)
}
func TestUncoveredRangesRightOnly(t *testing.T) {
req := rangeOf(t, "2026-01-01 00:00", "2026-01-31 00:00")
cov := rangeOf(t, "2026-01-01 00:00", "2026-01-15 00:00")
want := rangeOf(t, "2026-01-15 00:00", "2026-01-31 00:00")
got := uncoveredRanges(req.start, req.end, cov.start, cov.end, true)
assertRanges(t, got, want)
}
func TestUncoveredRangesBothSides(t *testing.T) {
req := rangeOf(t, "2026-01-01 00:00", "2026-01-31 00:00")
cov := rangeOf(t, "2026-01-10 00:00", "2026-01-20 00:00")
wantLeft := rangeOf(t, "2026-01-01 00:00", "2026-01-10 00:00")
wantRight := rangeOf(t, "2026-01-20 00:00", "2026-01-31 00:00")
got := uncoveredRanges(req.start, req.end, cov.start, cov.end, true)
assertRanges(t, got, wantLeft, wantRight)
}

View file

@ -18,6 +18,7 @@ import (
log "github.com/nacorid/logger"
"github.com/nacorid/naco-api/internal/activitypub"
"github.com/nacorid/naco-api/internal/bluesky"
"github.com/nacorid/naco-api/internal/clustering"
"github.com/nacorid/naco-api/internal/config"
"github.com/nacorid/naco-api/internal/nostr"
"github.com/nacorid/naco-api/internal/ollama"
@ -30,23 +31,28 @@ import (
)
type Server struct {
listenAddress string
permissions os.FileMode
mcpHandler *x402.X402Server
db *store.Queries
apClient apBackend
bskyClient blueskyBackend
nostrClient nostrBackend
twitterClient twitterBackend
ollama ollamaBackend
metrics *MetricsMiddleware
postCache *ttlCache
prewarmAuthors []string
prewarmFeedURI string
enrichEnabled bool
x402Wallet string
activeNetworks []NetworkInfo
enrichSem chan struct{}
listenAddress string
permissions os.FileMode
mcpHandler *x402.X402Server
db *store.Queries
apClient apBackend
bskyClient blueskyBackend
nostrClient nostrBackend
twitterClient twitterBackend
ollama ollamaBackend
metrics *MetricsMiddleware
postCache *ttlCache
prewarmAuthors []string
prewarmFeedURI string
enrichEnabled bool
x402Wallet string
activeNetworks []NetworkInfo
enrichSem chan struct{}
analyticsTTLEngagement time.Duration
analyticsTTLPatterns time.Duration
analyticsTTLTopics time.Duration
analyticsTTLSentiment time.Duration
analyticsTTLDrivers time.Duration
}
func New(ctx context.Context, db *store.Queries, ap *activitypub.APClient, bsky *bluesky.BlueskyClient, n *nostr.Client, t *twitter.Client, o *ollama.OllamaClient, cfg *config.Config) *Server {
@ -110,7 +116,7 @@ func New(ctx context.Context, db *store.Queries, ap *activitypub.APClient, bsky
txHash := settleCtx.Result.Transaction
platform := "unknown"
payload := map[string]any{
"event": "transaction_success",
"event": "transaction_success",
"transaction": map[string]any{
"x402_transaction_hash": txHash,
},
@ -141,23 +147,28 @@ func New(ctx context.Context, db *store.Queries, ap *activitypub.APClient, bsky
}
s := &Server{
listenAddress: cfg.ListenAddress,
permissions: cfg.UnixSocketPermissions,
mcpHandler: srv,
db: db,
apClient: ap,
bskyClient: bsky,
nostrClient: n,
twitterClient: t,
ollama: o,
metrics: metrics,
postCache: newTTLCache(postCacheTTL),
prewarmAuthors: cfg.PrewarmAuthors,
prewarmFeedURI: cfg.PrewarmFeedURI,
enrichEnabled: cfg.EnrichCrossPlatform,
x402Wallet: cfg.X402WalletAddress,
activeNetworks: activeNetworks,
enrichSem: make(chan struct{}, 4),
listenAddress: cfg.ListenAddress,
permissions: cfg.UnixSocketPermissions,
mcpHandler: srv,
db: db,
apClient: ap,
bskyClient: bsky,
nostrClient: n,
twitterClient: t,
ollama: o,
metrics: metrics,
postCache: newTTLCache(postCacheTTL),
prewarmAuthors: cfg.PrewarmAuthors,
prewarmFeedURI: cfg.PrewarmFeedURI,
enrichEnabled: cfg.EnrichCrossPlatform,
x402Wallet: cfg.X402WalletAddress,
activeNetworks: activeNetworks,
enrichSem: make(chan struct{}, 4),
analyticsTTLEngagement: cfg.AnalyticsTTLEngagement,
analyticsTTLPatterns: cfg.AnalyticsTTLPatterns,
analyticsTTLTopics: cfg.AnalyticsTTLTopics,
analyticsTTLSentiment: cfg.AnalyticsTTLSentiment,
analyticsTTLDrivers: cfg.AnalyticsTTLDrivers,
}
s.registerTools()
@ -463,6 +474,171 @@ func (s *Server) registerTools() {
s.buildRequirements("0.01", extra)...,
)
// --- Analytics tools ---
extra = map[string]any{
"OutputSchema": Schema{
Input: InputSchema{
Type: "HTTP",
Method: "POST",
Discoverable: true,
BodyFields: map[string]any{
"author": Bodyfield{Type: "string", Description: "Did or handle of an author", Required: true},
"startTime": Bodyfield{Type: "string", Description: "Start time in RFC3339 format (default: 30 days ago)", Required: false},
"endTime": Bodyfield{Type: "string", Description: "End time in RFC3339 format (default: now)", Required: false},
"limit": Bodyfield{Type: "integer", Description: "Maximum number of posts to analyze (default: 50, max: 200)", Required: false},
"platform": Bodyfield{Type: "string", Description: "Platform to filter (default: all)", Required: false},
},
},
Output: map[string]any{
"author": Bodyfield{Type: "string", Description: "Author identifier"},
"metrics": Bodyfield{Type: "array", Description: "Engagement metrics per post"},
},
},
}
s.mcpHandler.AddPayableTool(
mcp.Tool{Name: "getEngagementMetrics",
Description: "Get time-series engagement metrics (likes, reposts, replies, quotes) per post",
InputSchema: AnalyticsRequest{},
OutputSchema: EngagementMetricsResponse{},
},
s.getEngagementMetrics,
s.buildRequirements("0.15", extra)...,
)
extra = map[string]any{
"OutputSchema": Schema{
Input: InputSchema{
Type: "HTTP",
Method: "POST",
Discoverable: true,
BodyFields: map[string]any{
"author": Bodyfield{Type: "string", Description: "Did or handle of an author", Required: true},
"startTime": Bodyfield{Type: "string", Description: "Start time in RFC3339 format (default: 30 days ago)", Required: false},
"endTime": Bodyfield{Type: "string", Description: "End time in RFC3339 format (default: now)", Required: false},
"limit": Bodyfield{Type: "integer", Description: "Maximum number of posts to analyze (default: 50, max: 200)", Required: false},
"platform": Bodyfield{Type: "string", Description: "Platform to filter (default: all)", Required: false},
},
},
Output: map[string]any{
"author": Bodyfield{Type: "string", Description: "Author identifier"},
"heatmap": Bodyfield{Type: "array", Description: "24x7 matrix: hour (0-23) x day of week (0-6 = Sun-Sat)"},
"peakHours": Bodyfield{Type: "array", Description: "Top 3 hours with most posts (0-23)"},
"peakDays": Bodyfield{Type: "array", Description: "Top 3 days with most posts (0-6 = Sun-Sat)"},
"avgPostsPerDay": Bodyfield{Type: "number", Description: "Average posts per day"},
"totalPosts": Bodyfield{Type: "integer", Description: "Total posts in time range"},
"postingDays": Bodyfield{Type: "integer", Description: "Number of days with at least one post"},
},
},
}
s.mcpHandler.AddPayableTool(
mcp.Tool{Name: "getPostingPatterns",
Description: "Get posting heatmap and patterns (hour-of-day x day-of-week)",
InputSchema: AnalyticsRequest{},
OutputSchema: PostingPatternsResponse{},
},
s.getPostingPatterns,
s.buildRequirements("0.10", extra)...,
)
extra = map[string]any{
"OutputSchema": Schema{
Input: InputSchema{
Type: "HTTP",
Method: "POST",
Discoverable: true,
BodyFields: map[string]any{
"author": Bodyfield{Type: "string", Description: "Did or handle of an author", Required: true},
"startTime": Bodyfield{Type: "string", Description: "Start time in RFC3339 format (default: 30 days ago)", Required: false},
"endTime": Bodyfield{Type: "string", Description: "End time in RFC3339 format (default: now)", Required: false},
"limit": Bodyfield{Type: "integer", Description: "Maximum number of posts to analyze (default: 50, max: 200)", Required: false},
"platform": Bodyfield{Type: "string", Description: "Platform to filter (default: all)", Required: false},
},
},
Output: map[string]any{
"author": Bodyfield{Type: "string", Description: "Author identifier"},
"topics": Bodyfield{Type: "array", Description: "Identified topic clusters"},
},
},
}
s.mcpHandler.AddPayableTool(
mcp.Tool{Name: "getTopicModeling",
Description: "Get LLM-based topic extraction and clustering from posts",
InputSchema: AnalyticsRequest{},
OutputSchema: TopicModelingResponse{},
},
s.getTopicModeling,
s.buildRequirements("0.20", extra)...,
)
extra = map[string]any{
"OutputSchema": Schema{
Input: InputSchema{
Type: "HTTP",
Method: "POST",
Discoverable: true,
BodyFields: map[string]any{
"author": Bodyfield{Type: "string", Description: "Did or handle of an author", Required: true},
"startTime": Bodyfield{Type: "string", Description: "Start time in RFC3339 format (default: 30 days ago)", Required: false},
"endTime": Bodyfield{Type: "string", Description: "End time in RFC3339 format (default: now)", Required: false},
"limit": Bodyfield{Type: "integer", Description: "Maximum number of posts to analyze (default: 50, max: 200)", Required: false},
"platform": Bodyfield{Type: "string", Description: "Platform to filter (default: all)", Required: false},
},
},
Output: map[string]any{
"author": Bodyfield{Type: "string", Description: "Author identifier"},
"timeSeries": Bodyfield{Type: "array", Description: "Sentiment over time"},
"overallSentiment": Bodyfield{Type: "object", Description: "Overall sentiment summary"},
"dominantEmotions": Bodyfield{Type: "array", Description: "Most frequent emotions"},
"sentimentShifts": Bodyfield{Type: "array", Description: "Significant sentiment changes"},
},
},
}
s.mcpHandler.AddPayableTool(
mcp.Tool{Name: "getSentimentTrends",
Description: "Get sentiment analysis and emotional arcs over time",
InputSchema: AnalyticsRequest{},
OutputSchema: SentimentTrendsResponse{},
},
s.getSentimentTrends,
s.buildRequirements("0.15", extra)...,
)
extra = map[string]any{
"OutputSchema": Schema{
Input: InputSchema{
Type: "HTTP",
Method: "POST",
Discoverable: true,
BodyFields: map[string]any{
"author": Bodyfield{Type: "string", Description: "Did or handle of an author", Required: true},
"startTime": Bodyfield{Type: "string", Description: "Start time in RFC3339 format (default: 30 days ago)", Required: false},
"endTime": Bodyfield{Type: "string", Description: "End time in RFC3339 format (default: now)", Required: false},
"limit": Bodyfield{Type: "integer", Description: "Maximum number of posts to analyze (default: 50, max: 200)", Required: false},
"platform": Bodyfield{Type: "string", Description: "Platform to filter (default: all)", Required: false},
},
},
Output: map[string]any{
"author": Bodyfield{Type: "string", Description: "Author identifier"},
"drivers": Bodyfield{Type: "array", Description: "Topics/keywords driving engagement"},
},
},
}
s.mcpHandler.AddPayableTool(
mcp.Tool{Name: "getEngagementDrivers",
Description: "Get topics/keywords that correlate with high engagement",
InputSchema: AnalyticsRequest{},
OutputSchema: EngagementDriversResponse{},
},
s.getEngagementDrivers,
s.buildRequirements("0.20", extra)...,
)
s.mcpHandler.AddTool(
mcp.Tool{Name: "getFeed",
Description: "Get a feed by its URI",
@ -1269,6 +1445,22 @@ func (s *Server) buildStyleProfile(ctx context.Context, platform platform, autho
return nil, fmt.Errorf("no posts found for author %s", author)
}
// Delete old posts for non-Twitter platforms before rebuild
if platform != Twitter {
if err := s.db.DeleteStylePosts(ctx, store.DeleteStylePostsParams{
UserID: authorID,
Platform: platform.String(),
}); err != nil {
log.WarnWithContext(ctx, "failed to delete old style posts", "author", authorID, "platform", platform, "error", err)
}
}
// Extract texts for embedding and keyword analysis
postTexts := make([]string, len(posts))
for i, p := range posts {
postTexts[i] = p.Text
}
// Fail fast on any server-side error. These are payable tools: returning a degraded
// profile here would bill the caller for a deliverable they didn't ask for. The x402
// layer only settles payment when the handler succeeds, so returning the error means
@ -1278,7 +1470,7 @@ func (s *Server) buildStyleProfile(ctx context.Context, platform platform, autho
return nil, err
}
personaEmbedding := utils.AverageVector(embeddings)
topWords := utils.TopWords(posts, make(map[string]struct{}), 10)
topWords := utils.TopWords(postTexts, make(map[string]struct{}), 10)
styleSummary, err := s.ollama.SummarizeStyle(ctx, posts)
if err != nil {
return nil, err
@ -1389,61 +1581,109 @@ func (s *Server) prewarmOne(ctx context.Context, author string) {
log.InfoWithContext(ctx, "prewarmed style profile", "author", author)
}
func (s *Server) fetchProfilePosts(ctx context.Context, platform platform, author string, limit int) ([]string, error) {
func (s *Server) fetchProfilePosts(ctx context.Context, platform platform, author string, limit int) ([]clustering.Item, error) {
platformStr := platform.String()
// 1. Try cache first
cached, err := s.db.GetStylePosts(ctx, store.GetStylePostsParams{
UserID: author,
Platform: platformStr,
Limit: int32(limit),
})
if err == nil && len(cached) >= limit {
posts := make([]clustering.Item, len(cached))
for i, c := range cached {
posts[i] = clustering.Item{ID: c.PostID, Text: c.PostText}
}
return posts, nil
}
// 2. Fetch from platform
var freshPosts []clustering.Item
switch platform {
case Bluesky:
res, err := s.bskyClient.GetLatestNPosts(ctx, author, limit)
if err != nil {
return nil, err
return toItems(cached), err
}
posts := make([]string, 0, len(res))
freshPosts = make([]clustering.Item, 0, len(res))
for _, p := range res {
fp, ok := p.Post.Record.Val.(*bsky.FeedPost)
if !ok {
continue
}
posts = append(posts, fp.Text)
freshPosts = append(freshPosts, clustering.Item{
ID: p.Post.Uri,
Text: fp.Text,
})
}
return posts, nil
case ActivityPub:
res, err := s.apClient.GetLatestNPosts(ctx, author, limit)
if err != nil {
return nil, err
return toItems(cached), err
}
posts := make([]string, 0, len(res))
freshPosts = make([]clustering.Item, 0, len(res))
for _, p := range res {
if p.Text != "" {
posts = append(posts, p.Text)
freshPosts = append(freshPosts, clustering.Item{
ID: p.ID,
Text: p.Text,
})
}
}
return posts, nil
case Nostr:
res, err := s.nostrClient.GetLatestNPosts(ctx, author, limit)
if err != nil {
return nil, err
return toItems(cached), err
}
posts := make([]string, 0, len(res))
freshPosts = make([]clustering.Item, 0, len(res))
for _, p := range res {
if p.Content != "" {
posts = append(posts, p.Content)
freshPosts = append(freshPosts, clustering.Item{
ID: p.ID,
Text: p.Content,
})
}
}
return posts, nil
case Twitter:
res, err := s.twitterClient.GetLatestNPosts(ctx, author, limit)
if err != nil {
return nil, err
return toItems(cached), err
}
posts := make([]string, 0, len(res))
freshPosts = make([]clustering.Item, 0, len(res))
for _, p := range res {
if p.Content != "" {
posts = append(posts, p.Content)
freshPosts = append(freshPosts, clustering.Item{
ID: p.ID,
Text: p.Content,
})
}
}
return posts, nil
default:
return nil, fmt.Errorf("platform %q is not supported yet", platform)
}
// 3. Upsert to cache (batch)
for _, post := range freshPosts {
if err := s.db.UpsertStylePosts(ctx, store.UpsertStylePostsParams{
UserID: author,
Platform: platformStr,
PostID: post.ID,
PostText: post.Text,
}); err != nil {
log.WarnWithContext(ctx, "failed to cache style post", "author", author, "platform", platform, "postID", post.ID, "error", err)
}
}
return freshPosts, nil
}
func toItems(rows []store.GetStylePostsRow) []clustering.Item {
items := make([]clustering.Item, len(rows))
for i, r := range rows {
items[i] = clustering.Item{ID: r.PostID, Text: r.PostText}
}
return items
}
func determinePlatform(author string) (platform, error) {
@ -1552,5 +1792,16 @@ func parseActiveNetworks(raw string) []string {
}
func (s *Server) Start() error {
go s.startPostPruner()
return s.mcpHandler.Start(s.listenAddress, s.permissions)
}
func (s *Server) startPostPruner() {
ticker := time.NewTicker(24 * time.Hour)
defer ticker.Stop()
for range ticker.C {
if err := s.db.PruneStylePosts(context.Background()); err != nil {
log.Error("failed to prune style posts", "error", err)
}
}
}

View file

@ -151,3 +151,99 @@ func platformFromString(name string) (platform, bool) {
return -1, false
}
}
// --- Analytics types ---
type AnalyticsRequest struct {
Author string `json:"author" jsonschema_description:"Did or handle of an author"`
StartTime string `json:"startTime,omitempty" jsonschema_description:"Start time in RFC3339 format (default: 30 days ago)"`
EndTime string `json:"endTime,omitempty" jsonschema_description:"End time in RFC3339 format (default: now)"`
Limit int `json:"limit,omitempty" jsonschema_description:"Maximum number of posts to analyze (default: 50, max: 200)"`
Platform string `json:"platform,omitempty" jsonschema_description:"Platform to filter (default: all)"`
}
type EngagementMetricsResponse struct {
Author string `json:"author" jsonschema_description:"Author identifier"`
Metrics []EngagementMetricPoint `json:"metrics" jsonschema_description:"Engagement metrics per post"`
}
type EngagementMetricPoint struct {
PostID string `json:"postId" jsonschema_description:"Post identifier"`
Timestamp string `json:"timestamp" jsonschema_description:"RFC3339 timestamp"`
Likes int64 `json:"likes" jsonschema_description:"Number of likes"`
Reposts int64 `json:"reposts" jsonschema_description:"Number of reposts"`
Replies int64 `json:"replies" jsonschema_description:"Number of replies"`
Quotes int64 `json:"quotes" jsonschema_description:"Number of quotes"`
EngagementRate float64 `json:"engagementRate" jsonschema_description:"Engagement rate (interactions / impressions estimate)"`
}
type PostingPatternsResponse struct {
Author string `json:"author" jsonschema_description:"Author identifier"`
Heatmap [][]int `json:"heatmap" jsonschema_description:"24x7 matrix: hour (0-23) x day of week (0-6 = Sun-Sat)"`
PeakHours []int `json:"peakHours" jsonschema_description:"Top 3 hours with most posts (0-23)"`
PeakDays []int `json:"peakDays" jsonschema_description:"Top 3 days with most posts (0-6 = Sun-Sat)"`
AvgPostsPerDay float64 `json:"avgPostsPerDay" jsonschema_description:"Average posts per day"`
TotalPosts int `json:"totalPosts" jsonschema_description:"Total posts in time range"`
PostingDays int `json:"postingDays" jsonschema_description:"Number of days with at least one post"`
}
type TopicModelingResponse struct {
Author string `json:"author" jsonschema_description:"Author identifier"`
Topics []TopicCluster `json:"topics" jsonschema_description:"Identified topic clusters"`
}
type TopicCluster struct {
TopicID int `json:"topicId" jsonschema_description:"Cluster identifier"`
Label string `json:"label" jsonschema_description:"Human-readable topic label"`
Keywords []string `json:"keywords" jsonschema_description:"Top keywords for this topic"`
PostCount int `json:"postCount" jsonschema_description:"Number of posts in this topic"`
Percentage float64 `json:"percentage" jsonschema_description:"Percentage of total posts"`
SamplePosts []string `json:"samplePosts" jsonschema_description:"Sample post texts (max 3)"`
}
type SentimentTrendsResponse struct {
Author string `json:"author" jsonschema_description:"Author identifier"`
TimeSeries []SentimentPoint `json:"timeSeries" jsonschema_description:"Sentiment over time"`
OverallSentiment SentimentSummary `json:"overallSentiment" jsonschema_description:"Overall sentiment summary"`
DominantEmotions []EmotionScore `json:"dominantEmotions" jsonschema_description:"Most frequent emotions"`
SentimentShifts []SentimentShift `json:"sentimentShifts" jsonschema_description:"Significant sentiment changes"`
}
type SentimentPoint struct {
Timestamp string `json:"timestamp" jsonschema_description:"RFC3339 timestamp (bucketed by day)"`
Positive float64 `json:"positive" jsonschema_description:"Positive sentiment ratio (0-1)"`
Negative float64 `json:"negative" jsonschema_description:"Negative sentiment ratio (0-1)"`
Neutral float64 `json:"neutral" jsonschema_description:"Neutral sentiment ratio (0-1)"`
PostCount int `json:"postCount" jsonschema_description:"Posts in this time bucket"`
}
type SentimentSummary struct {
Positive float64 `json:"positive" jsonschema_description:"Overall positive ratio"`
Negative float64 `json:"negative" jsonschema_description:"Overall negative ratio"`
Neutral float64 `json:"neutral" jsonschema_description:"Overall neutral ratio"`
}
type EmotionScore struct {
Emotion string `json:"emotion" jsonschema_description:"Emotion name"`
Score float64 `json:"score" jsonschema_description:"Frequency score (0-1)"`
}
type SentimentShift struct {
Timestamp string `json:"timestamp" jsonschema_description:"When the shift occurred"`
FromSentiment string `json:"fromSentiment" jsonschema_description:"Previous dominant sentiment"`
ToSentiment string `json:"toSentiment" jsonschema_description:"New dominant sentiment"`
Magnitude float64 `json:"magnitude" jsonschema_description:"Shift magnitude (0-1)"`
}
type EngagementDriversResponse struct {
Author string `json:"author" jsonschema_description:"Author identifier"`
Drivers []EngagementDriver `json:"drivers" jsonschema_description:"Topics/keywords driving engagement"`
}
type EngagementDriver struct {
Topic string `json:"topic" jsonschema_description:"Topic or keyword"`
AvgEngagement float64 `json:"avgEngagement" jsonschema_description:"Average engagement rate for posts with this topic"`
Correlation float64 `json:"correlation" jsonschema_description:"Correlation coefficient with engagement (-1 to 1)"`
PostCount int `json:"postCount" jsonschema_description:"Number of posts containing this topic"`
SamplePosts []string `json:"samplePosts" jsonschema_description:"Sample posts (max 2)"`
}

View file

@ -25,7 +25,7 @@ func newWebhookSender(cfg *config.Config, m *MetricsMiddleware) *WebhookSender {
}
return &WebhookSender{
cfg: cfg,
cfg: cfg,
metrics: m,
httpCli: &http.Client{Timeout: 10 * time.Second},
server: cfg.MatrixServerID,
@ -38,7 +38,7 @@ func (wh *WebhookSender) Send(ctx context.Context, jsonBytes []byte) {
return
}
if jsonBytes == nil || len(jsonBytes) == 0 {
if len(jsonBytes) == 0 {
log.ErrorWithContext(ctx, "webhook: empty payload ignored")
return
}

View file

@ -0,0 +1,96 @@
// Code generated by sqlc. DO NOT EDIT.
// versions:
// sqlc v1.31.1
// source: AnalyticsCache.sql
package store
import (
"context"
"encoding/json"
"time"
)
const deleteAnalyticsCache = `-- name: DeleteAnalyticsCache :exec
DELETE FROM analytics_cache
WHERE user_id = $1 AND platform = $2 AND tool_name = $3 AND time_range_hash = $4
`
type DeleteAnalyticsCacheParams struct {
UserID string
Platform string
ToolName string
TimeRangeHash string
}
func (q *Queries) DeleteAnalyticsCache(ctx context.Context, arg DeleteAnalyticsCacheParams) error {
_, err := q.db.ExecContext(ctx, deleteAnalyticsCache,
arg.UserID,
arg.Platform,
arg.ToolName,
arg.TimeRangeHash,
)
return err
}
const getAnalyticsCache = `-- name: GetAnalyticsCache :one
SELECT result FROM analytics_cache
WHERE user_id = $1 AND platform = $2 AND tool_name = $3 AND time_range_hash = $4
AND expires_at > now()
`
type GetAnalyticsCacheParams struct {
UserID string
Platform string
ToolName string
TimeRangeHash string
}
func (q *Queries) GetAnalyticsCache(ctx context.Context, arg GetAnalyticsCacheParams) (json.RawMessage, error) {
row := q.db.QueryRowContext(ctx, getAnalyticsCache,
arg.UserID,
arg.Platform,
arg.ToolName,
arg.TimeRangeHash,
)
var result json.RawMessage
err := row.Scan(&result)
return result, err
}
const pruneAnalyticsCache = `-- name: PruneAnalyticsCache :exec
DELETE FROM analytics_cache WHERE expires_at <= now()
`
func (q *Queries) PruneAnalyticsCache(ctx context.Context) error {
_, err := q.db.ExecContext(ctx, pruneAnalyticsCache)
return err
}
const upsertAnalyticsCache = `-- name: UpsertAnalyticsCache :exec
INSERT INTO analytics_cache (user_id, platform, tool_name, time_range_hash, result, expires_at)
VALUES ($1, $2, $3, $4, $5, $6)
ON CONFLICT (user_id, platform, tool_name, time_range_hash)
DO UPDATE SET result = EXCLUDED.result, expires_at = EXCLUDED.expires_at, computed_at = now()
`
type UpsertAnalyticsCacheParams struct {
UserID string
Platform string
ToolName string
TimeRangeHash string
Result json.RawMessage
ExpiresAt time.Time
}
func (q *Queries) UpsertAnalyticsCache(ctx context.Context, arg UpsertAnalyticsCacheParams) error {
_, err := q.db.ExecContext(ctx, upsertAnalyticsCache,
arg.UserID,
arg.Platform,
arg.ToolName,
arg.TimeRangeHash,
arg.Result,
arg.ExpiresAt,
)
return err
}

View file

@ -0,0 +1,59 @@
// Code generated by sqlc. DO NOT EDIT.
// versions:
// sqlc v1.31.1
// source: AnalyticsCoverage.sql
package store
import (
"context"
"time"
)
const getAnalyticsCoverage = `-- name: GetAnalyticsCoverage :one
SELECT covered_start, covered_end FROM analytics_coverage
WHERE user_id = $1 AND platform = $2
`
type GetAnalyticsCoverageParams struct {
UserID string
Platform string
}
type GetAnalyticsCoverageRow struct {
CoveredStart time.Time
CoveredEnd time.Time
}
func (q *Queries) GetAnalyticsCoverage(ctx context.Context, arg GetAnalyticsCoverageParams) (GetAnalyticsCoverageRow, error) {
row := q.db.QueryRowContext(ctx, getAnalyticsCoverage, arg.UserID, arg.Platform)
var i GetAnalyticsCoverageRow
err := row.Scan(&i.CoveredStart, &i.CoveredEnd)
return i, err
}
const upsertAnalyticsCoverage = `-- name: UpsertAnalyticsCoverage :exec
INSERT INTO analytics_coverage (user_id, platform, covered_start, covered_end, updated_at)
VALUES ($1, $2, $3, $4, now())
ON CONFLICT (user_id, platform)
DO UPDATE SET covered_start = LEAST(analytics_coverage.covered_start, EXCLUDED.covered_start),
covered_end = GREATEST(analytics_coverage.covered_end, EXCLUDED.covered_end),
updated_at = now()
`
type UpsertAnalyticsCoverageParams struct {
UserID string
Platform string
CoveredStart time.Time
CoveredEnd time.Time
}
func (q *Queries) UpsertAnalyticsCoverage(ctx context.Context, arg UpsertAnalyticsCoverageParams) error {
_, err := q.db.ExecContext(ctx, upsertAnalyticsCoverage,
arg.UserID,
arg.Platform,
arg.CoveredStart,
arg.CoveredEnd,
)
return err
}

View file

@ -0,0 +1,113 @@
// Code generated by sqlc. DO NOT EDIT.
// versions:
// sqlc v1.31.1
// source: AnalyticsPosts.sql
package store
import (
"context"
"time"
)
const getAnalyticsPostsInRange = `-- name: GetAnalyticsPostsInRange :many
SELECT post_id, post_text, post_timestamp, likes, reposts, replies, quotes
FROM analytics_posts
WHERE user_id = $1 AND platform = $2 AND post_timestamp >= $3 AND post_timestamp <= $4
ORDER BY post_timestamp DESC
LIMIT $5
`
type GetAnalyticsPostsInRangeParams struct {
UserID string
Platform string
PostTimestamp time.Time
PostTimestamp_2 time.Time
Limit int32
}
type GetAnalyticsPostsInRangeRow struct {
PostID string
PostText string
PostTimestamp time.Time
Likes int64
Reposts int64
Replies int64
Quotes int64
}
func (q *Queries) GetAnalyticsPostsInRange(ctx context.Context, arg GetAnalyticsPostsInRangeParams) ([]GetAnalyticsPostsInRangeRow, error) {
rows, err := q.db.QueryContext(ctx, getAnalyticsPostsInRange,
arg.UserID,
arg.Platform,
arg.PostTimestamp,
arg.PostTimestamp_2,
arg.Limit,
)
if err != nil {
return nil, err
}
defer rows.Close()
var items []GetAnalyticsPostsInRangeRow
for rows.Next() {
var i GetAnalyticsPostsInRangeRow
if err := rows.Scan(
&i.PostID,
&i.PostText,
&i.PostTimestamp,
&i.Likes,
&i.Reposts,
&i.Replies,
&i.Quotes,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Close(); err != nil {
return nil, err
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const upsertAnalyticsPosts = `-- name: UpsertAnalyticsPosts :exec
INSERT INTO analytics_posts (user_id, platform, post_id, post_text, post_timestamp, likes, reposts, replies, quotes)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
ON CONFLICT (user_id, platform, post_id)
DO UPDATE SET post_text = EXCLUDED.post_text,
post_timestamp = EXCLUDED.post_timestamp,
likes = EXCLUDED.likes,
reposts = EXCLUDED.reposts,
replies = EXCLUDED.replies,
quotes = EXCLUDED.quotes
`
type UpsertAnalyticsPostsParams struct {
UserID string
Platform string
PostID string
PostText string
PostTimestamp time.Time
Likes int64
Reposts int64
Replies int64
Quotes int64
}
func (q *Queries) UpsertAnalyticsPosts(ctx context.Context, arg UpsertAnalyticsPostsParams) error {
_, err := q.db.ExecContext(ctx, upsertAnalyticsPosts,
arg.UserID,
arg.Platform,
arg.PostID,
arg.PostText,
arg.PostTimestamp,
arg.Likes,
arg.Reposts,
arg.Replies,
arg.Quotes,
)
return err
}

View file

@ -0,0 +1,25 @@
// Code generated by sqlc. DO NOT EDIT.
// versions:
// sqlc v1.31.1
// source: DeleteStylePosts.sql
package store
import (
"context"
)
const deleteStylePosts = `-- name: DeleteStylePosts :exec
DELETE FROM user_style_posts
WHERE user_id = $1 AND platform = $2 AND platform != 'Twitter'
`
type DeleteStylePostsParams struct {
UserID string
Platform string
}
func (q *Queries) DeleteStylePosts(ctx context.Context, arg DeleteStylePostsParams) error {
_, err := q.db.ExecContext(ctx, deleteStylePosts, arg.UserID, arg.Platform)
return err
}

View file

@ -20,14 +20,14 @@ LIMIT $2
`
type FindSimilarUsersParams struct {
QueryEmbedding any
QueryEmbedding interface{}
LimitVal int32
}
type FindSimilarUsersRow struct {
UserID string
Platform string
Embedding any
Embedding interface{}
TopKeywords []string
StyleSummary string
PostCount int32

View file

@ -0,0 +1,33 @@
// Code generated by sqlc. DO NOT EDIT.
// versions:
// sqlc v1.31.1
// source: GetStylePostByID.sql
package store
import (
"context"
)
const getStylePostByID = `-- name: GetStylePostByID :one
SELECT post_id, post_text FROM user_style_posts
WHERE user_id = $1 AND platform = $2 AND post_id = $3
`
type GetStylePostByIDParams struct {
UserID string
Platform string
PostID string
}
type GetStylePostByIDRow struct {
PostID string
PostText string
}
func (q *Queries) GetStylePostByID(ctx context.Context, arg GetStylePostByIDParams) (GetStylePostByIDRow, error) {
row := q.db.QueryRowContext(ctx, getStylePostByID, arg.UserID, arg.Platform, arg.PostID)
var i GetStylePostByIDRow
err := row.Scan(&i.PostID, &i.PostText)
return i, err
}

View file

@ -0,0 +1,51 @@
// Code generated by sqlc. DO NOT EDIT.
// versions:
// sqlc v1.31.1
// source: GetStylePosts.sql
package store
import (
"context"
)
const getStylePosts = `-- name: GetStylePosts :many
SELECT post_id, post_text FROM user_style_posts
WHERE user_id = $1 AND platform = $2
ORDER BY created_at DESC
LIMIT $3
`
type GetStylePostsParams struct {
UserID string
Platform string
Limit int32
}
type GetStylePostsRow struct {
PostID string
PostText string
}
func (q *Queries) GetStylePosts(ctx context.Context, arg GetStylePostsParams) ([]GetStylePostsRow, error) {
rows, err := q.db.QueryContext(ctx, getStylePosts, arg.UserID, arg.Platform, arg.Limit)
if err != nil {
return nil, err
}
defer rows.Close()
var items []GetStylePostsRow
for rows.Next() {
var i GetStylePostsRow
if err := rows.Scan(&i.PostID, &i.PostText); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Close(); err != nil {
return nil, err
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}

View file

@ -0,0 +1,51 @@
// Code generated by sqlc. DO NOT EDIT.
// versions:
// sqlc v1.31.1
// source: GetStylePostsByIDs.sql
package store
import (
"context"
"github.com/lib/pq"
)
const getStylePostsByIDs = `-- name: GetStylePostsByIDs :many
SELECT post_id, post_text FROM user_style_posts
WHERE user_id = $1 AND platform = $2 AND post_id = ANY($3::text[])
`
type GetStylePostsByIDsParams struct {
UserID string
Platform string
Column3 []string
}
type GetStylePostsByIDsRow struct {
PostID string
PostText string
}
func (q *Queries) GetStylePostsByIDs(ctx context.Context, arg GetStylePostsByIDsParams) ([]GetStylePostsByIDsRow, error) {
rows, err := q.db.QueryContext(ctx, getStylePostsByIDs, arg.UserID, arg.Platform, pq.Array(arg.Column3))
if err != nil {
return nil, err
}
defer rows.Close()
var items []GetStylePostsByIDsRow
for rows.Next() {
var i GetStylePostsByIDsRow
if err := rows.Scan(&i.PostID, &i.PostText); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Close(); err != nil {
return nil, err
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}

View file

@ -0,0 +1,20 @@
// Code generated by sqlc. DO NOT EDIT.
// versions:
// sqlc v1.31.1
// source: PruneStylePosts.sql
package store
import (
"context"
)
const pruneStylePosts = `-- name: PruneStylePosts :exec
DELETE FROM user_style_posts
WHERE created_at < now() - interval '7 days' AND platform != 'Twitter'
`
func (q *Queries) PruneStylePosts(ctx context.Context) error {
_, err := q.db.ExecContext(ctx, pruneStylePosts)
return err
}

View file

@ -0,0 +1,34 @@
// Code generated by sqlc. DO NOT EDIT.
// versions:
// sqlc v1.31.1
// source: UpsertStylePosts.sql
package store
import (
"context"
)
const upsertStylePosts = `-- name: UpsertStylePosts :exec
INSERT INTO user_style_posts (user_id, platform, post_id, post_text)
VALUES ($1, $2, $3, $4)
ON CONFLICT (user_id, platform, post_id)
DO UPDATE SET post_text = EXCLUDED.post_text, created_at = now()
`
type UpsertStylePostsParams struct {
UserID string
Platform string
PostID string
PostText string
}
func (q *Queries) UpsertStylePosts(ctx context.Context, arg UpsertStylePostsParams) error {
_, err := q.db.ExecContext(ctx, upsertStylePosts,
arg.UserID,
arg.Platform,
arg.PostID,
arg.PostText,
)
return err
}

View file

@ -26,7 +26,7 @@ DO UPDATE SET
type UpsertStyleProfileParams struct {
UserID string
Platform string
Embedding any
Embedding interface{}
TopKeywords []string
StyleSummary string
PostCount int32

View file

@ -10,10 +10,10 @@ import (
)
type DBTX interface {
ExecContext(context.Context, string, ...any) (sql.Result, error)
ExecContext(context.Context, string, ...interface{}) (sql.Result, error)
PrepareContext(context.Context, string) (*sql.Stmt, error)
QueryContext(context.Context, string, ...any) (*sql.Rows, error)
QueryRowContext(context.Context, string, ...any) *sql.Row
QueryContext(context.Context, string, ...interface{}) (*sql.Rows, error)
QueryRowContext(context.Context, string, ...interface{}) *sql.Row
}
func New(db DBTX) *Queries {

View file

@ -35,7 +35,7 @@ func TestStoreRoundTrip(t *testing.T) {
q := New(conn)
uid := fmt.Sprintf("test_%d", time.Now().UnixNano())
emb := "[" + strings.TrimSuffix(strings.Repeat("0.0,", 768), ",") + "]"
emb := "[" + strings.TrimSuffix(strings.Repeat("0.0,", 1024), ",") + "]"
if err := q.UpsertStyleProfile(ctx, UpsertStyleProfileParams{
UserID: uid,

View file

@ -1,13 +1,48 @@
// Code generated by sqlc. DO NOT EDIT.
// versions:
// sqlc v1.30.0
// sqlc v1.31.1
package store
import (
"database/sql"
"encoding/json"
"time"
)
type AnalyticsCache struct {
ID int64
UserID string
Platform string
ToolName string
TimeRangeHash string
Result json.RawMessage
ComputedAt time.Time
ExpiresAt time.Time
}
type AnalyticsCoverage struct {
UserID string
Platform string
CoveredStart time.Time
CoveredEnd time.Time
UpdatedAt time.Time
}
type AnalyticsPost struct {
ID int64
UserID string
Platform string
PostID string
PostText string
PostTimestamp time.Time
Likes int64
Reposts int64
Replies int64
Quotes int64
CreatedAt time.Time
}
type Transaction struct {
ID int64
X402TransactionHash string
@ -19,11 +54,20 @@ type Transaction struct {
CreatedAt sql.NullTime
}
type UserStylePost struct {
ID int64
UserID string
Platform string
PostID string
PostText string
CreatedAt time.Time
}
type UserStyleProfile struct {
ID int64
UserID string
Platform string
Embedding any
Embedding interface{}
TopKeywords []string
StyleSummary string
PostCount int32

View file

@ -38,11 +38,9 @@ func ParseEmbedding(value any) ([]float64, error) {
return vec, nil
case string:
// pgvector returns a string like "[0.12, 0.53, ...]" sometimes
return parseVectorString(v)
case []byte:
// lib/pq returns unknown-type columns (e.g. vector) as []byte
return parseVectorString(string(v))
default:
@ -76,7 +74,7 @@ func AverageVector(vs [][]float64) []float64 {
func TopWords(posts []string, stopWords map[string]struct{}, limit int) []string {
counts := map[string]int{}
for _, p := range posts {
for _, w := range strings.Fields(strings.ToLower(p)) {
for w := range strings.FieldsSeq(strings.ToLower(p)) {
if len(w) < 4 {
continue
}

62
scripts/fund-wallet.sh Executable file
View file

@ -0,0 +1,62 @@
#!/bin/bash
set -euo pipefail
echo "=== Funding Wallet with Base Sepolia USDC ==="
# Required environment variables
: "${PRIVATE_KEY:?PRIVATE_KEY must be set}"
: "${FUNDING_FAUCET_URL:?FUNDING_FAUCET_URL must be set}"
WALLET_ADDRESS=$(cast wallet address --private-key "$PRIVATE_KEY")
echo "Wallet address: $WALLET_ADDRESS"
# Check current USDC balance
USDC_CONTRACT="0x036CbD53842c5426634e7929541eC2318f3dCF7e"
RPC_URL="https://sepolia.base.org"
BALANCE=$(cast call "$USDC_CONTRACT" "balanceOf(address)(uint256)" "$WALLET_ADDRESS" --rpc-url "$RPC_URL" 2>/dev/null || echo "0")
BALANCE_DECIMAL=$(cast to-dec "$BALANCE" 2>/dev/null || echo "0")
BALANCE_USDC=$(echo "scale=6; $BALANCE_DECIMAL / 1000000" | bc 2>/dev/null || echo "0")
echo "Current USDC balance: $BALANCE_USDC USDC"
# If balance is less than 10 USDC, request from faucet
if (( $(echo "$BALANCE_USDC < 10" | bc -l) )); then
echo "Balance below 10 USDC, requesting from faucet..."
# Request from faucet (Coinbase CDP faucet API format)
curl -X POST "$FUNDING_FAUCET_URL" \
-H "Content-Type: application/json" \
-d "{\"address\": \"$WALLET_ADDRESS\", \"token\": \"USDC\"}" \
|| echo "Faucet request failed, may need manual funding"
# Wait for funding
echo "Waiting for funding..."
sleep 30
# Check balance again
BALANCE=$(cast call "$USDC_CONTRACT" "balanceOf(address)(uint256)" "$WALLET_ADDRESS" --rpc-url "$RPC_URL" 2>/dev/null || echo "0")
BALANCE_DECIMAL=$(cast to-dec "$BALANCE" 2>/dev/null || echo "0")
BALANCE_USDC=$(echo "scale=6; $BALANCE_DECIMAL / 1000000" | bc 2>/dev/null || echo "0")
echo "New USDC balance: $BALANCE_USDC USDC"
else
echo "Sufficient USDC balance"
fi
# Also fund with ETH for gas
ETH_BALANCE=$(cast balance "$WALLET_ADDRESS" --rpc-url "$RPC_URL" 2>/dev/null || echo "0")
ETH_BALANCE_DECIMAL=$(cast to-dec "$ETH_BALANCE" 2>/dev/null || echo "0")
ETH_BALANCE_ETH=$(echo "scale=6; $ETH_BALANCE_DECIMAL / 1000000000000000000" | bc 2>/dev/null || echo "0")
echo "Current ETH balance: $ETH_BALANCE_ETH ETH"
if (( $(echo "$ETH_BALANCE_ETH < 0.01" | bc -l) )); then
echo "ETH balance low, requesting from faucet..."
curl -X POST "$FUNDING_FAUCET_URL" \
-H "Content-Type: application/json" \
-d "{\"address\": \"$WALLET_ADDRESS\", \"token\": \"ETH\"}" \
|| echo "ETH faucet request failed"
sleep 10
fi
echo "=== Wallet Funding Complete ==="

83
scripts/start-test-server.sh Executable file
View file

@ -0,0 +1,83 @@
#!/bin/bash
set -euo pipefail
echo "=== Starting Test Server ==="
# Required environment variables
: "${PRIVATE_KEY:?PRIVATE_KEY must be set}"
: "${X402_WALLET_ADDRESS:?X402_WALLET_ADDRESS must be set}"
# Set up environment for local testing
export SERVER_ADDRESS="127.0.0.1"
export SERVER_PORT="8080"
export GOOSE_DBSTRING="postgres://postgres:postgres@localhost:5432/mcpserver?sslmode=disable"
export ACTIVE_NETWORKS="sepolia"
export X402_FACILITATOR_URL="https://api.cdp.coinbase.com/platform/v2/x402"
export X402_WALLET_ADDRESS="$X402_WALLET_ADDRESS"
export OLLAMA_HOST="http://localhost:11434"
export MIGRATE_ON_START="true"
export MIGRATIONS_DIR="sql/schema"
export READYZ_PORT="9090"
export METRICS_PORT="9091"
export LOG_FILE="/tmp/mcpserver-test.log"
export LOG_LEVEL="debug"
# Use test credentials or empty for local testing
export BLUESKY_DID="${BLUESKY_DID:-}"
export BLUESKY_APP_PASSWORD="${BLUESKY_APP_PASSWORD:-}"
export BLUESKY_HOST="${BLUESKY_HOST:-https://bsky.social}"
export AP_HOST="${AP_HOST:-https://mastodon.social}"
export AP_ACCESS_TOKEN="${AP_ACCESS_TOKEN:-}"
export NOSTR_RELAYS="${NOSTR_RELAYS:-wss://relay.damus.io,wss://nos.lol}"
export TWITTER_BEARER_TOKEN="${TWITTER_BEARER_TOKEN:-}"
export CDP_API_KEY_ID="${CDP_API_KEY_ID:-}"
export CDP_API_SECRET="${CDP_API_SECRET:-}"
export ENRICH_CROSS_PLATFORM="false"
export PREWARM_AUTHORS=""
export PREWARM_FEED_URI=""
# Start the server in background
echo "Starting server on port 8080..."
go run ./cmd/server/main.go &
SERVER_PID=$!
# Save PID for cleanup
echo $SERVER_PID > /tmp/test-server.pid
# Wait for server to be ready
echo "Waiting for server to be ready..."
for i in {1..30}; do
if curl -sf http://localhost:9090/readyz >/dev/null 2>&1; then
echo "Server is ready"
break
fi
sleep 1
done
# Verify server is responding
curl -sf http://localhost:8080/healthz >/dev/null 2>&1 || {
echo "Server health check failed"
kill $SERVER_PID 2>/dev/null || true
exit 1
}
# Verify payment endpoints are working (should return 402 without payment)
echo "Verifying payment endpoints..."
if curl -sf -X POST http://localhost:8080/v1/getActor \
-H "Content-Type: application/json" \
-d '{"author": "did:plc:z72i7hdynmk6r22z27h6tvur"}' >/dev/null 2>&1; then
echo "WARNING: Payable endpoint getActor returned 200 without payment!"
else
echo "Payment endpoint getActor correctly requires payment (402)"
fi
# Verify free endpoint works
if curl -sf -X POST http://localhost:8080/v1/getFeed \
-H "Content-Type: application/json" \
-d '{"uri": "https://bsky.app/profile/did:plc:z72i7hdynmk6r22z27h6tvur/feed/whats-hot"}' >/dev/null 2>&1; then
echo "Free endpoint getFeed works correctly"
else
echo "WARNING: Free endpoint getFeed failed!"
fi
echo "=== Test Server Started (PID: $SERVER_PID) ==="

View file

@ -0,0 +1,17 @@
-- name: GetAnalyticsCache :one
SELECT result FROM analytics_cache
WHERE user_id = $1 AND platform = $2 AND tool_name = $3 AND time_range_hash = $4
AND expires_at > now();
-- name: UpsertAnalyticsCache :exec
INSERT INTO analytics_cache (user_id, platform, tool_name, time_range_hash, result, expires_at)
VALUES ($1, $2, $3, $4, $5, $6)
ON CONFLICT (user_id, platform, tool_name, time_range_hash)
DO UPDATE SET result = EXCLUDED.result, expires_at = EXCLUDED.expires_at, computed_at = now();
-- name: DeleteAnalyticsCache :exec
DELETE FROM analytics_cache
WHERE user_id = $1 AND platform = $2 AND tool_name = $3 AND time_range_hash = $4;
-- name: PruneAnalyticsCache :exec
DELETE FROM analytics_cache WHERE expires_at <= now();

View file

@ -0,0 +1,11 @@
-- name: GetAnalyticsCoverage :one
SELECT covered_start, covered_end FROM analytics_coverage
WHERE user_id = $1 AND platform = $2;
-- name: UpsertAnalyticsCoverage :exec
INSERT INTO analytics_coverage (user_id, platform, covered_start, covered_end, updated_at)
VALUES ($1, $2, $3, $4, now())
ON CONFLICT (user_id, platform)
DO UPDATE SET covered_start = LEAST(analytics_coverage.covered_start, EXCLUDED.covered_start),
covered_end = GREATEST(analytics_coverage.covered_end, EXCLUDED.covered_end),
updated_at = now();

View file

@ -0,0 +1,17 @@
-- name: GetAnalyticsPostsInRange :many
SELECT post_id, post_text, post_timestamp, likes, reposts, replies, quotes
FROM analytics_posts
WHERE user_id = $1 AND platform = $2 AND post_timestamp >= $3 AND post_timestamp <= $4
ORDER BY post_timestamp DESC
LIMIT $5;
-- name: UpsertAnalyticsPosts :exec
INSERT INTO analytics_posts (user_id, platform, post_id, post_text, post_timestamp, likes, reposts, replies, quotes)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
ON CONFLICT (user_id, platform, post_id)
DO UPDATE SET post_text = EXCLUDED.post_text,
post_timestamp = EXCLUDED.post_timestamp,
likes = EXCLUDED.likes,
reposts = EXCLUDED.reposts,
replies = EXCLUDED.replies,
quotes = EXCLUDED.quotes;

View file

@ -0,0 +1,3 @@
-- name: DeleteStylePosts :exec
DELETE FROM user_style_posts
WHERE user_id = $1 AND platform = $2 AND platform != 'Twitter';

View file

@ -0,0 +1,3 @@
-- name: GetStylePostByID :one
SELECT post_id, post_text FROM user_style_posts
WHERE user_id = $1 AND platform = $2 AND post_id = $3;

View file

@ -0,0 +1,5 @@
-- name: GetStylePosts :many
SELECT post_id, post_text FROM user_style_posts
WHERE user_id = $1 AND platform = $2
ORDER BY created_at DESC
LIMIT $3;

View file

@ -0,0 +1,3 @@
-- name: GetStylePostsByIDs :many
SELECT post_id, post_text FROM user_style_posts
WHERE user_id = $1 AND platform = $2 AND post_id = ANY($3::text[]);

View file

@ -0,0 +1,3 @@
-- name: PruneStylePosts :exec
DELETE FROM user_style_posts
WHERE created_at < now() - interval '7 days' AND platform != 'Twitter';

View file

@ -0,0 +1,5 @@
-- name: UpsertStylePosts :exec
INSERT INTO user_style_posts (user_id, platform, post_id, post_text)
VALUES ($1, $2, $3, $4)
ON CONFLICT (user_id, platform, post_id)
DO UPDATE SET post_text = EXCLUDED.post_text, created_at = now();

View file

@ -1,5 +1,5 @@
-- +goose up
CREATE TABLE transactions (
CREATE TABLE IF NOT EXISTS transactions (
id BIGSERIAL PRIMARY KEY,
x402_transaction_hash VARCHAR(255) NOT NULL UNIQUE,
amount NUMERIC NOT NULL,

View file

@ -1,11 +1,11 @@
-- +goose up
CREATE EXTENSION IF NOT EXISTS vector;
CREATE TABLE user_style_profiles (
CREATE TABLE IF NOT EXISTS user_style_profiles (
id BIGSERIAL PRIMARY KEY,
user_id TEXT NOT NULL,
platform TEXT NOT NULL,
embedding VECTOR(768) NOT NULL,
embedding VECTOR(1024) NOT NULL,
top_keywords TEXT[] NOT NULL,
style_summary TEXT NOT NULL,
post_count INTEGER NOT NULL,

View file

@ -4,7 +4,7 @@ ALTER TABLE user_style_profiles
ADD COLUMN schema_version INTEGER NOT NULL DEFAULT 1,
ADD COLUMN algorithm_version INTEGER NOT NULL DEFAULT 1;
CREATE INDEX idx_user_style_profiles_versions
CREATE INDEX IF NOT EXISTS idx_user_style_profiles_versions
ON user_style_profiles (model_version, schema_version, algorithm_version);
-- +goose Down

View file

@ -0,0 +1,17 @@
-- +goose up
CREATE TABLE IF NOT EXISTS user_style_posts (
id BIGSERIAL PRIMARY KEY,
user_id TEXT NOT NULL,
platform TEXT NOT NULL,
post_id TEXT NOT NULL,
post_text TEXT NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
UNIQUE (user_id, platform, post_id)
);
CREATE INDEX idx_user_style_posts_user_platform
ON user_style_posts (user_id, platform);
-- +goose down
DROP INDEX IF EXISTS idx_user_style_posts_user_platform;
DROP TABLE IF EXISTS user_style_posts;

View file

@ -0,0 +1,23 @@
-- +goose up
CREATE TABLE IF NOT EXISTS analytics_cache (
id BIGSERIAL PRIMARY KEY,
user_id TEXT NOT NULL,
platform TEXT NOT NULL,
tool_name TEXT NOT NULL,
time_range_hash TEXT NOT NULL,
result JSONB NOT NULL,
computed_at TIMESTAMPTZ NOT NULL DEFAULT now(),
expires_at TIMESTAMPTZ NOT NULL,
UNIQUE (user_id, platform, tool_name, time_range_hash)
);
CREATE INDEX IF NOT EXISTS idx_analytics_cache_lookup
ON analytics_cache (user_id, platform, tool_name, time_range_hash);
CREATE INDEX IF NOT EXISTS idx_analytics_cache_expires
ON analytics_cache (expires_at);
-- +goose down
DROP INDEX IF EXISTS idx_analytics_cache_expires;
DROP INDEX IF EXISTS idx_analytics_cache_lookup;
DROP TABLE IF EXISTS analytics_cache;

View file

@ -0,0 +1,22 @@
-- +goose up
CREATE TABLE IF NOT EXISTS analytics_posts (
id BIGSERIAL PRIMARY KEY,
user_id TEXT NOT NULL,
platform TEXT NOT NULL,
post_id TEXT NOT NULL,
post_text TEXT NOT NULL,
post_timestamp TIMESTAMPTZ NOT NULL,
likes BIGINT NOT NULL DEFAULT 0,
reposts BIGINT NOT NULL DEFAULT 0,
replies BIGINT NOT NULL DEFAULT 0,
quotes BIGINT NOT NULL DEFAULT 0,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
UNIQUE (user_id, platform, post_id)
);
CREATE INDEX IF NOT EXISTS idx_analytics_posts_user_platform_time
ON analytics_posts (user_id, platform, post_timestamp);
-- +goose down
DROP INDEX IF EXISTS idx_analytics_posts_user_platform_time;
DROP TABLE IF EXISTS analytics_posts;

View file

@ -0,0 +1,12 @@
-- +goose up
CREATE TABLE IF NOT EXISTS analytics_coverage (
user_id TEXT NOT NULL,
platform TEXT NOT NULL,
covered_start TIMESTAMPTZ NOT NULL,
covered_end TIMESTAMPTZ NOT NULL,
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
PRIMARY KEY (user_id, platform)
);
-- +goose down
DROP TABLE IF EXISTS analytics_coverage;

View file

@ -1,16 +1,22 @@
package main
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"log"
"os"
"math/big"
"net/http"
"net/http/httputil"
"os"
"strings"
"time"
"github.com/ethereum/go-ethereum"
"github.com/ethereum/go-ethereum/accounts/abi"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/ethclient"
"github.com/joho/godotenv"
x402 "github.com/x402-foundation/x402/go/v2"
@ -19,6 +25,15 @@ import (
evmsigners "github.com/x402-foundation/x402/go/v2/signers/evm"
)
const (
USDCContractAddress = "0x036CbD53842c5426634e7929541eC2318f3dCF7e"
BaseSepoliaRPC = "https://sepolia.base.org"
MinUSDCRequired = 10.0
MinETHRequired = 0.01
)
var usdcABI = `[{"constant":true,"inputs":[{"name":"_owner","type":"address"}],"name":"balanceOf","outputs":[{"name":"balance","type":"uint256"}],"type":"function"}]`
type LoggingRoundTripper struct {
Proxied http.RoundTripper
}
@ -31,7 +46,6 @@ func (lrt *LoggingRoundTripper) RoundTrip(req *http.Request) (*http.Response, er
req.Body = rc
}
// Dump request
reqDump, err := httputil.DumpRequestOut(req, true)
if err != nil {
log.Printf("Error dumping request: %v", err)
@ -39,13 +53,11 @@ func (lrt *LoggingRoundTripper) RoundTrip(req *http.Request) (*http.Response, er
fmt.Printf("\n--- REQUEST ---\n%s\n----------------\n", reqDump)
}
// Perform request
resp, err := lrt.Proxied.RoundTrip(req)
if err != nil {
return nil, err
}
// Dump response
respDump, err := httputil.DumpResponse(resp, true)
if err != nil {
log.Printf("Error dumping response: %v", err)
@ -56,13 +68,206 @@ func (lrt *LoggingRoundTripper) RoundTrip(req *http.Request) (*http.Response, er
return resp, nil
}
func callEndpoint(client *http.Client, method, endpoint, body string) error {
baseURL := getEnv("TEST_SERVER_URL", "http://localhost:8080")
url := fmt.Sprintf("%s/v1/%s", baseURL, endpoint)
req, err := http.NewRequest(method, url, strings.NewReader(body))
if err != nil {
return fmt.Errorf("failed to create %s request: %w", endpoint, err)
}
req.Header.Set("Content-Type", "application/json")
resp, err := client.Do(req)
if err != nil {
return fmt.Errorf("%s HTTP request failed: %w", endpoint, err)
}
defer resp.Body.Close()
respBody, _ := io.ReadAll(resp.Body)
fmt.Printf("%s Response Status: %s\n", endpoint, resp.Status)
if len(respBody) > 0 {
fmt.Printf("%s Response Body: %s\n", endpoint, string(respBody))
}
if resp.StatusCode == 402 {
return fmt.Errorf("%s: payment required (402)", endpoint)
}
return nil
}
func getEnv(key, fallback string) string {
if value, ok := os.LookupEnv(key); ok {
return value
}
return fallback
}
func getUSDCBalance(client *ethclient.Client, address common.Address) (*big.Int, error) {
contractAddr := common.HexToAddress(USDCContractAddress)
parsedABI, err := abi.JSON(strings.NewReader(usdcABI))
if err != nil {
return nil, fmt.Errorf("failed to parse USDC ABI: %w", err)
}
data, err := parsedABI.Pack("balanceOf", address)
if err != nil {
return nil, fmt.Errorf("failed to pack balanceOf call: %w", err)
}
msg := ethereum.CallMsg{
To: &contractAddr,
Data: data,
}
result, err := client.CallContract(context.Background(), msg, nil)
if err != nil {
return nil, fmt.Errorf("failed to call balanceOf: %w", err)
}
unpacked, err := parsedABI.Unpack("balanceOf", result)
if err != nil {
return nil, fmt.Errorf("failed to unpack balance: %w", err)
}
if len(unpacked) == 0 {
return nil, fmt.Errorf("no balance returned")
}
balance, ok := unpacked[0].(*big.Int)
if !ok {
return nil, fmt.Errorf("unexpected balance type: %T", unpacked[0])
}
return balance, nil
}
func getETHBalance(client *ethclient.Client, address common.Address) (*big.Int, error) {
balance, err := client.BalanceAt(context.Background(), address, nil)
if err != nil {
return nil, fmt.Errorf("failed to get ETH balance: %w", err)
}
return balance, nil
}
func requestFaucetFunds(faucetURL, address, token string) error {
payload := map[string]string{
"address": address,
"token": token,
}
jsonData, err := json.Marshal(payload)
if err != nil {
return fmt.Errorf("failed to marshal faucet request: %w", err)
}
resp, err := http.Post(faucetURL, "application/json", strings.NewReader(string(jsonData)))
if err != nil {
return fmt.Errorf("faucet request failed: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode >= 400 {
body, _ := io.ReadAll(resp.Body)
return fmt.Errorf("faucet returned error %d: %s", resp.StatusCode, string(body))
}
return nil
}
func ensureFundedWallet(privateKey, faucetURL, rpcURL string) error {
client, err := ethclient.Dial(rpcURL)
if err != nil {
return fmt.Errorf("failed to connect to RPC: %w", err)
}
defer client.Close()
signer, err := evmsigners.NewClientSignerFromPrivateKey(privateKey)
if err != nil {
return fmt.Errorf("failed to create signer: %w", err)
}
addrStr := signer.Address()
address := common.HexToAddress(addrStr)
log.Printf("Checking wallet balance for %s...", addrStr)
usdcBalance, err := getUSDCBalance(client, address)
if err != nil {
return fmt.Errorf("failed to get USDC balance: %w", err)
}
usdcBalanceFloat := new(big.Float).Quo(new(big.Float).SetInt(usdcBalance), big.NewFloat(1e6))
usdcFloat, _ := usdcBalanceFloat.Float64()
log.Printf("Current USDC balance: %.6f USDC", usdcFloat)
ethBalance, err := getETHBalance(client, address)
if err != nil {
return fmt.Errorf("failed to get ETH balance: %w", err)
}
ethBalanceFloat := new(big.Float).Quo(new(big.Float).SetInt(ethBalance), big.NewFloat(1e18))
ethFloat, _ := ethBalanceFloat.Float64()
log.Printf("Current ETH balance: %.6f ETH", ethFloat)
needsFunding := false
if usdcFloat < MinUSDCRequired {
log.Printf("USDC balance %.6f below minimum %.2f, requesting funding...", usdcFloat, MinUSDCRequired)
needsFunding = true
}
if ethFloat < MinETHRequired {
log.Printf("ETH balance %.6f below minimum %.4f, requesting funding...", ethFloat, MinETHRequired)
needsFunding = true
}
if needsFunding {
if faucetURL == "" {
return fmt.Errorf("FUNDING_FAUCET_URL not set but wallet needs funding")
}
if usdcFloat < MinUSDCRequired {
log.Printf("Requesting USDC from faucet...")
if err := requestFaucetFunds(faucetURL, addrStr, "USDC"); err != nil {
return fmt.Errorf("failed to request USDC from faucet: %w", err)
}
}
if ethFloat < MinETHRequired {
log.Printf("Requesting ETH from faucet...")
if err := requestFaucetFunds(faucetURL, addrStr, "ETH"); err != nil {
return fmt.Errorf("failed to request ETH from faucet: %w", err)
}
}
log.Printf("Waiting for funding to arrive...")
time.Sleep(30 * time.Second)
usdcBalance, err = getUSDCBalance(client, address)
if err != nil {
return fmt.Errorf("failed to get USDC balance after funding: %w", err)
}
usdcBalanceFloat = new(big.Float).Quo(new(big.Float).SetInt(usdcBalance), big.NewFloat(1e6))
usdcFloat, _ = usdcBalanceFloat.Float64()
log.Printf("New USDC balance: %.6f USDC", usdcFloat)
ethBalance, err = getETHBalance(client, address)
if err != nil {
return fmt.Errorf("failed to get ETH balance after funding: %w", err)
}
ethBalanceFloat = new(big.Float).Quo(new(big.Float).SetInt(ethBalance), big.NewFloat(1e18))
ethFloat, _ = ethBalanceFloat.Float64()
log.Printf("New ETH balance: %.6f ETH", ethFloat)
if usdcFloat < MinUSDCRequired {
return fmt.Errorf("USDC balance still insufficient after funding: %.6f < %.2f", usdcFloat, MinUSDCRequired)
}
if ethFloat < MinETHRequired {
return fmt.Errorf("ETH balance still insufficient after funding: %.6f < %.4f", ethFloat, MinETHRequired)
}
} else {
log.Printf("Wallet has sufficient funds")
}
return nil
}
func main() {
godotenv.Load()
var (
privateKey = os.Getenv("PRIVATE_KEY")
serverURL = "https://api.naco.li/v1"
ctx = context.Background()
)
privateKey := os.Getenv("PRIVATE_KEY")
serverURL := getEnv("TEST_SERVER_URL", "http://localhost:8080")
faucetURL := getEnv("FUNDING_FAUCET_URL", "")
rpcURL := getEnv("RPC_URL", BaseSepoliaRPC)
if privateKey == "" {
log.Fatal("PRIVATE_KEY environment variable is required")
}
log.Printf("Ensuring wallet is funded...")
if err := ensureFundedWallet(privateKey, faucetURL, rpcURL); err != nil {
log.Fatal("Failed to ensure wallet is funded: ", err)
}
signer, err := evmsigners.NewClientSignerFromPrivateKey(privateKey)
if err != nil {
@ -70,12 +275,13 @@ func main() {
}
x402Client := x402.Newx402Client().
Register("eip155:*", evm.NewExactEvmScheme(signer, nil))
Register("eip155:84532", evm.NewExactEvmScheme(signer, nil))
debugClient := &http.Client{
Transport: &LoggingRoundTripper{
Proxied: http.DefaultTransport,
},
Timeout: 60 * time.Second,
}
httpClient := x402http.WrapHTTPClientWithPayment(
@ -83,39 +289,44 @@ func main() {
x402http.Newx402HTTPClient(x402Client),
)
log.Println("Starting REST API tests...")
log.Printf("Starting REST API endpoint tests against %s...", serverURL)
log.Println("\nCalling getFeed (free tool) via HTTP client...")
bodyGetFeed := []byte(`{"uri": "https://bsky.app/profile/did:plc:z72i7hdynmk6r22z27h6tvur/feed/whats-hot"}`)
resp, err := httpClient.Post(fmt.Sprintf("%s/getFeed", serverURL), "application/json", bytes.NewReader(bodyGetFeed))
if err != nil {
log.Println("getFeed HTTP request failed:", err)
} else {
defer resp.Body.Close()
body, _ := io.ReadAll(resp.Body)
log.Printf("getFeed Response Status: %s\n", resp.Status)
log.Printf("getFeed Response Body: %s\n", string(body))
testCases := []struct {
name string
method string
endpoint string
body string
}{
{"getFeed", "POST", "getFeed", `{"uri": "https://bsky.app/profile/did:plc:z72i7hdynmk6r22z27h6tvur/feed/whats-hot"}`},
{"getActor", "POST", "getActor", `{"author": "did:plc:z72i7hdynmk6r22z27h6tvur"}`},
{"searchActors", "POST", "searchActors", `{"query": "test", "limit": 5}`},
{"getPost", "POST", "getPost", `{"actor":"sharkgamiart.bsky.social","rkey":"3m47y27c67c2b"}`},
{"getLatestPosts", "POST", "getLatestPosts", `{"author":"sharkgamiart.bsky.social","limit": 5}`},
{"getPostsInTime", "POST", "getPostsInTime", `{"author":"sharkgamiart.bsky.social","startTime": "2024-01-01T00:00:00Z","endTime": "2024-12-31T23:59:59Z","limit": 5}`},
{"getStyleProfile", "POST", "getStyleProfile", `{"author":"did:plc:z72i7hdynmk6r22z27h6tvur"}`},
{"findSimilarUsers", "POST", "findSimilarUsers", `{"author":"did:plc:z72i7hdynmk6r22z27h6tvur","limit": 5}`},
}
log.Println("\nCalling getPost (gated paid tool) via HTTP client...")
bodyGetPost := []byte(`{"actor":"sharkgamiart.bsky.social","rkey":"3m47y27c67c2b"}`)
passed := 0
failed := 0
req, err := http.NewRequestWithContext(ctx, "POST", fmt.Sprintf("%s/getPost", serverURL), bytes.NewReader(bodyGetPost))
if err != nil {
log.Fatal("Failed to create getPost request:", err)
for _, tc := range testCases {
fmt.Printf("\n=== Testing %s ===\n", tc.name)
err := callEndpoint(httpClient, tc.method, tc.endpoint, tc.body)
if err != nil {
log.Printf("FAIL %s: %v\n", tc.name, err)
failed++
} else {
fmt.Printf("PASS %s\n", tc.name)
passed++
}
}
req.GetBody = func() (io.ReadCloser, error) {
return io.NopCloser(bytes.NewReader(bodyGetPost)), nil
}
req.Header.Set("Content-Type", "application/json")
respPost, err := httpClient.Do(req)
if err != nil {
log.Println("getPost HTTP request failed:", err)
} else {
defer respPost.Body.Close()
body, _ := io.ReadAll(respPost.Body)
log.Printf("getPost Response Status: %s\n", respPost.Status)
log.Printf("getPost Response Body: %s\n", string(body))
log.Printf("\n=== Test Summary ===")
log.Printf("Passed: %d", passed)
log.Printf("Failed: %d", failed)
if failed > 0 {
os.Exit(1)
}
}