nacorid/naco-api
Watch
1
0
Fork
You've already forked naco-api
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Compare commits

base: nacorid:2ba8f5cbef081a5bc42976ed2bec619471c98f3c
Branches Tags
nacorid:main
nacorid:ai-main
..
compare: nacorid:f8efe2f551dbcb649f0efa97ca584870a4090542
Branches Tags
nacorid:ai-main
nacorid:main
54 changed files with 143 additions and 3547 deletions
Download patch file Download diff file Expand all files Collapse all files

142
.forgejo/workflows/test-endpoints.yaml
View file

@ -1,142 +0,0 @@
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,5 +1,3 @@
.env .env
cdpScripts/cdp cdpScripts/cdp
bin/ bin/
test-endpoints
test-endpoints.exe

2
cmd/server/main.go
View file

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

5
config/models.env
View file

@ -1,5 +0,0 @@
# 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
View file

@ -1,26 +0,0 @@
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/deckarep/golang-set/v2 v2.8.0 // indirect
github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.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/c-kzg-4844/v2 v2.1.6 // indirect
github.com/ethereum/go-ethereum v1.17.2 github.com/ethereum/go-ethereum v1.17.2 // indirect
github.com/holiman/uint256 v1.3.2 // indirect github.com/holiman/uint256 v1.3.2 // indirect
github.com/klauspost/compress v1.18.1 // indirect github.com/klauspost/compress v1.18.1 // indirect
github.com/supranational/blst v0.3.16 // indirect github.com/supranational/blst v0.3.16 // indirect

24
internal/activitypub/types.go
View file

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

37
internal/clustering/helpers.go
View file

@ -1,37 +0,0 @@
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
}

228
internal/clustering/k-means.go
View file

@ -1,228 +0,0 @@
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
}

273
internal/clustering/kmeans_test.go
View file

@ -1,273 +0,0 @@
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))
}
}
}

21
internal/clustering/types.go
View file

@ -1,21 +0,0 @@
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
}

64
internal/config/config.go
View file

@ -2,10 +2,8 @@ package config
import ( import (
"os" "os"
"path/filepath"
"strconv" "strconv"
"strings" "strings"
"time"
"github.com/joho/godotenv" "github.com/joho/godotenv"
log "github.com/nacorid/logger" log "github.com/nacorid/logger"
@ -40,32 +38,9 @@ type Config struct {
ActiveNetworks string ActiveNetworks string
MatrixWebhookURL string MatrixWebhookURL string
MatrixServerID 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 { 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 { if err := godotenv.Load(); err != nil {
log.Debug("No .env file found, reading from environment") log.Debug("No .env file found, reading from environment")
} }
@ -91,7 +66,7 @@ func Load() *Config {
PrewarmFeedURI: getEnv("PREWARM_FEED_URI", ""), PrewarmFeedURI: getEnv("PREWARM_FEED_URI", ""),
EnrichCrossPlatform: getEnvBool("ENRICH_CROSS_PLATFORM", "true"), EnrichCrossPlatform: getEnvBool("ENRICH_CROSS_PLATFORM", "true"),
OllamaHost: getEnv("OLLAMA_HOST", "http://localhost:11434"), OllamaHost: getEnv("OLLAMA_HOST", "http://localhost:11434"),
MigrateOnStart: getEnvBool("MIGRATE_ON_START", "true"), MigrateOnStart: getEnvBool("MIGRATE_ON_START", "false"),
MigrationsDir: getEnv("MIGRATIONS_DIR", "sql/schema"), MigrationsDir: getEnv("MIGRATIONS_DIR", "sql/schema"),
ReadyzPort: getEnv("READYZ_PORT", "9090"), ReadyzPort: getEnv("READYZ_PORT", "9090"),
MetricsPort: getEnv("METRICS_PORT", "9091"), MetricsPort: getEnv("METRICS_PORT", "9091"),
@ -103,13 +78,6 @@ func Load() *Config {
ActiveNetworks: getEnv("ACTIVE_NETWORKS", "base"), ActiveNetworks: getEnv("ACTIVE_NETWORKS", "base"),
MatrixWebhookURL: getEnv("MATRIX_WEBHOOK_URL", ""), MatrixWebhookURL: getEnv("MATRIX_WEBHOOK_URL", ""),
MatrixServerID: getEnv("MATRIX_SERVER_ID", ""), 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 // Default MatrixServerID to system hostname if not explicitly set
@ -143,40 +111,12 @@ 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 { func splitComma(s string) []string {
parts := make([]string, 0) parts := make([]string, 0)
for p := range strings.SplitSeq(s, ",") { for _, p := range strings.Split(s, ",") {
if trimmed := strings.TrimSpace(p); trimmed != "" { if trimmed := strings.TrimSpace(p); trimmed != "" {
parts = append(parts, trimmed) parts = append(parts, trimmed)
} }
} }
return parts 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 "."
}

53
internal/ollama/client.go
View file

@ -6,64 +6,39 @@ import (
"net/url" "net/url"
"strings" "strings"
"github.com/nacorid/naco-api/internal/clustering"
"github.com/ollama/ollama/api" "github.com/ollama/ollama/api"
) )
type OllamaClient struct { type OllamaClient struct {
client api.Client client api.Client
embeddingModel string
styleModel string
} }
func New(ctx context.Context, host *url.URL, http *http.Client, em, sm string) *OllamaClient { func New(ctx context.Context, host *url.URL, http *http.Client) *OllamaClient {
client := api.NewClient(host, http) client := api.NewClient(host, http)
return &OllamaClient{client: *client, embeddingModel: em, styleModel: sm} return &OllamaClient{client: *client}
} }
func (o *OllamaClient) EmbedTexts(ctx context.Context, posts []clustering.Item) ([][]float64, error) { func (o *OllamaClient) EmbedTexts(ctx context.Context, posts []string) ([][]float64, error) {
if len(posts) == 0 { vectors := make([][]float64, 0, len(posts))
return nil, nil for _, t := range posts {
}
embeddings := make([][]float64, 0, len(posts))
for _, post := range posts {
resp, err := o.client.Embeddings(ctx, &api.EmbeddingRequest{ resp, err := o.client.Embeddings(ctx, &api.EmbeddingRequest{
Model: o.embeddingModel, Model: "nomic-embed-text",
Prompt: post.Text, Prompt: t,
}) })
if err != nil { if err != nil {
return nil, err return nil, err
} }
embeddings = append(embeddings, resp.Embedding) vectors = append(vectors, resp.Embedding)
} }
return vectors, nil
return embeddings, nil
} }
func (o *OllamaClient) SummarizeStyle(ctx context.Context, posts []clustering.Item) (string, error) { func (o *OllamaClient) SummarizeStyle(ctx context.Context, posts []string) (string, error) {
if len(posts) == 0 { prompt := "Summarize writing style, recurring themes, tone, personality traits of this user:\n\n" +
return "", nil strings.Join(posts, "\n") + "\n\nStyle summary:"
}
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 var fullResponse strings.Builder
err := o.client.Generate(ctx, &api.GenerateRequest{ err := o.client.Generate(ctx, &api.GenerateRequest{
Model: o.styleModel, Model: "mistral",
Prompt: prompt, Prompt: prompt,
}, func(resp api.GenerateResponse) error { }, func(resp api.GenerateResponse) error {
fullResponse.WriteString(resp.Response) fullResponse.WriteString(resp.Response)

260
internal/ollama/client_test.go
View file

@ -1,260 +0,0 @@
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)
}
}

883
internal/server/analytics.go
View file

@ -1,883 +0,0 @@
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
}

5
internal/server/backends.go
View file

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

71
internal/server/coverage_test.go
View file

@ -1,71 +0,0 @@
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)
}

357
internal/server/server.go
View file

@ -18,7 +18,6 @@ import (
log "github.com/nacorid/logger" log "github.com/nacorid/logger"
"github.com/nacorid/naco-api/internal/activitypub" "github.com/nacorid/naco-api/internal/activitypub"
"github.com/nacorid/naco-api/internal/bluesky" "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/config"
"github.com/nacorid/naco-api/internal/nostr" "github.com/nacorid/naco-api/internal/nostr"
"github.com/nacorid/naco-api/internal/ollama" "github.com/nacorid/naco-api/internal/ollama"
@ -31,28 +30,23 @@ import (
) )
type Server struct { type Server struct {
listenAddress string listenAddress string
permissions os.FileMode permissions os.FileMode
mcpHandler *x402.X402Server mcpHandler *x402.X402Server
db *store.Queries db *store.Queries
apClient apBackend apClient apBackend
bskyClient blueskyBackend bskyClient blueskyBackend
nostrClient nostrBackend nostrClient nostrBackend
twitterClient twitterBackend twitterClient twitterBackend
ollama ollamaBackend ollama ollamaBackend
metrics *MetricsMiddleware metrics *MetricsMiddleware
postCache *ttlCache postCache *ttlCache
prewarmAuthors []string prewarmAuthors []string
prewarmFeedURI string prewarmFeedURI string
enrichEnabled bool enrichEnabled bool
x402Wallet string x402Wallet string
activeNetworks []NetworkInfo activeNetworks []NetworkInfo
enrichSem chan struct{} 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 { 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 {
@ -116,7 +110,7 @@ func New(ctx context.Context, db *store.Queries, ap *activitypub.APClient, bsky
txHash := settleCtx.Result.Transaction txHash := settleCtx.Result.Transaction
platform := "unknown" platform := "unknown"
payload := map[string]any{ payload := map[string]any{
"event": "transaction_success", "event": "transaction_success",
"transaction": map[string]any{ "transaction": map[string]any{
"x402_transaction_hash": txHash, "x402_transaction_hash": txHash,
}, },
@ -147,28 +141,23 @@ func New(ctx context.Context, db *store.Queries, ap *activitypub.APClient, bsky
} }
s := &Server{ s := &Server{
listenAddress: cfg.ListenAddress, listenAddress: cfg.ListenAddress,
permissions: cfg.UnixSocketPermissions, permissions: cfg.UnixSocketPermissions,
mcpHandler: srv, mcpHandler: srv,
db: db, db: db,
apClient: ap, apClient: ap,
bskyClient: bsky, bskyClient: bsky,
nostrClient: n, nostrClient: n,
twitterClient: t, twitterClient: t,
ollama: o, ollama: o,
metrics: metrics, metrics: metrics,
postCache: newTTLCache(postCacheTTL), postCache: newTTLCache(postCacheTTL),
prewarmAuthors: cfg.PrewarmAuthors, prewarmAuthors: cfg.PrewarmAuthors,
prewarmFeedURI: cfg.PrewarmFeedURI, prewarmFeedURI: cfg.PrewarmFeedURI,
enrichEnabled: cfg.EnrichCrossPlatform, enrichEnabled: cfg.EnrichCrossPlatform,
x402Wallet: cfg.X402WalletAddress, x402Wallet: cfg.X402WalletAddress,
activeNetworks: activeNetworks, activeNetworks: activeNetworks,
enrichSem: make(chan struct{}, 4), enrichSem: make(chan struct{}, 4),
analyticsTTLEngagement: cfg.AnalyticsTTLEngagement,
analyticsTTLPatterns: cfg.AnalyticsTTLPatterns,
analyticsTTLTopics: cfg.AnalyticsTTLTopics,
analyticsTTLSentiment: cfg.AnalyticsTTLSentiment,
analyticsTTLDrivers: cfg.AnalyticsTTLDrivers,
} }
s.registerTools() s.registerTools()
@ -474,171 +463,6 @@ func (s *Server) registerTools() {
s.buildRequirements("0.01", extra)..., 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( s.mcpHandler.AddTool(
mcp.Tool{Name: "getFeed", mcp.Tool{Name: "getFeed",
Description: "Get a feed by its URI", Description: "Get a feed by its URI",
@ -1445,22 +1269,6 @@ func (s *Server) buildStyleProfile(ctx context.Context, platform platform, autho
return nil, fmt.Errorf("no posts found for author %s", author) 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 // 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 // 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 // layer only settles payment when the handler succeeds, so returning the error means
@ -1470,7 +1278,7 @@ func (s *Server) buildStyleProfile(ctx context.Context, platform platform, autho
return nil, err return nil, err
} }
personaEmbedding := utils.AverageVector(embeddings) personaEmbedding := utils.AverageVector(embeddings)
topWords := utils.TopWords(postTexts, make(map[string]struct{}), 10) topWords := utils.TopWords(posts, make(map[string]struct{}), 10)
styleSummary, err := s.ollama.SummarizeStyle(ctx, posts) styleSummary, err := s.ollama.SummarizeStyle(ctx, posts)
if err != nil { if err != nil {
return nil, err return nil, err
@ -1581,109 +1389,61 @@ func (s *Server) prewarmOne(ctx context.Context, author string) {
log.InfoWithContext(ctx, "prewarmed style profile", "author", author) log.InfoWithContext(ctx, "prewarmed style profile", "author", author)
} }
func (s *Server) fetchProfilePosts(ctx context.Context, platform platform, author string, limit int) ([]clustering.Item, error) { func (s *Server) fetchProfilePosts(ctx context.Context, platform platform, author string, limit int) ([]string, 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 { switch platform {
case Bluesky: case Bluesky:
res, err := s.bskyClient.GetLatestNPosts(ctx, author, limit) res, err := s.bskyClient.GetLatestNPosts(ctx, author, limit)
if err != nil { if err != nil {
return toItems(cached), err return nil, err
} }
freshPosts = make([]clustering.Item, 0, len(res)) posts := make([]string, 0, len(res))
for _, p := range res { for _, p := range res {
fp, ok := p.Post.Record.Val.(*bsky.FeedPost) fp, ok := p.Post.Record.Val.(*bsky.FeedPost)
if !ok { if !ok {
continue continue
} }
freshPosts = append(freshPosts, clustering.Item{ posts = append(posts, fp.Text)
ID: p.Post.Uri,
Text: fp.Text,
})
} }
return posts, nil
case ActivityPub: case ActivityPub:
res, err := s.apClient.GetLatestNPosts(ctx, author, limit) res, err := s.apClient.GetLatestNPosts(ctx, author, limit)
if err != nil { if err != nil {
return toItems(cached), err return nil, err
} }
freshPosts = make([]clustering.Item, 0, len(res)) posts := make([]string, 0, len(res))
for _, p := range res { for _, p := range res {
if p.Text != "" { if p.Text != "" {
freshPosts = append(freshPosts, clustering.Item{ posts = append(posts, p.Text)
ID: p.ID,
Text: p.Text,
})
} }
} }
return posts, nil
case Nostr: case Nostr:
res, err := s.nostrClient.GetLatestNPosts(ctx, author, limit) res, err := s.nostrClient.GetLatestNPosts(ctx, author, limit)
if err != nil { if err != nil {
return toItems(cached), err return nil, err
} }
freshPosts = make([]clustering.Item, 0, len(res)) posts := make([]string, 0, len(res))
for _, p := range res { for _, p := range res {
if p.Content != "" { if p.Content != "" {
freshPosts = append(freshPosts, clustering.Item{ posts = append(posts, p.Content)
ID: p.ID,
Text: p.Content,
})
} }
} }
return posts, nil
case Twitter: case Twitter:
res, err := s.twitterClient.GetLatestNPosts(ctx, author, limit) res, err := s.twitterClient.GetLatestNPosts(ctx, author, limit)
if err != nil { if err != nil {
return toItems(cached), err return nil, err
} }
freshPosts = make([]clustering.Item, 0, len(res)) posts := make([]string, 0, len(res))
for _, p := range res { for _, p := range res {
if p.Content != "" { if p.Content != "" {
freshPosts = append(freshPosts, clustering.Item{ posts = append(posts, p.Content)
ID: p.ID,
Text: p.Content,
})
} }
} }
return posts, nil
default: default:
return nil, fmt.Errorf("platform %q is not supported yet", platform) 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) { func determinePlatform(author string) (platform, error) {
@ -1792,16 +1552,5 @@ func parseActiveNetworks(raw string) []string {
} }
func (s *Server) Start() error { func (s *Server) Start() error {
go s.startPostPruner()
return s.mcpHandler.Start(s.listenAddress, s.permissions) 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)
}
}
}

96
internal/server/types.go
View file

@ -151,99 +151,3 @@ func platformFromString(name string) (platform, bool) {
return -1, false 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)"`
}

4
internal/server/webhook.go
View file

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

96
internal/store/AnalyticsCache.sql.go
View file

@ -1,96 +0,0 @@
// 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
}

59
internal/store/AnalyticsCoverage.sql.go
View file

@ -1,59 +0,0 @@
// 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
}

113
internal/store/AnalyticsPosts.sql.go
View file

@ -1,113 +0,0 @@
// 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
}

25
internal/store/DeleteStylePosts.sql.go
View file

@ -1,25 +0,0 @@
// 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
}

4
internal/store/FindSimilarUsers.sql.go
View file

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

33
internal/store/GetStylePostByID.sql.go
View file

@ -1,33 +0,0 @@
// 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
}

51
internal/store/GetStylePosts.sql.go
View file

@ -1,51 +0,0 @@
// 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
}

51
internal/store/GetStylePostsByIDs.sql.go
View file

@ -1,51 +0,0 @@
// 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
}

20
internal/store/PruneStylePosts.sql.go
View file

@ -1,20 +0,0 @@
// 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
}

34
internal/store/UpsertStylePosts.sql.go
View file

@ -1,34 +0,0 @@
// 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
}

2
internal/store/UpsertStyleProfiles.sql.go
View file

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

6
internal/store/db.go
View file

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

2
internal/store/integration_test.go
View file

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

48
internal/store/models.go
View file

@ -1,48 +1,13 @@
// Code generated by sqlc. DO NOT EDIT. // Code generated by sqlc. DO NOT EDIT.
// versions: // versions:
// sqlc v1.31.1 // sqlc v1.30.0
package store package store
import ( import (
"database/sql" "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 { type Transaction struct {
ID int64 ID int64
X402TransactionHash string X402TransactionHash string
@ -54,20 +19,11 @@ type Transaction struct {
CreatedAt sql.NullTime CreatedAt sql.NullTime
} }
type UserStylePost struct {
ID int64
UserID string
Platform string
PostID string
PostText string
CreatedAt time.Time
}
type UserStyleProfile struct { type UserStyleProfile struct {
ID int64 ID int64
UserID string UserID string
Platform string Platform string
Embedding interface{} Embedding any
TopKeywords []string TopKeywords []string
StyleSummary string StyleSummary string
PostCount int32 PostCount int32

4
internal/utils/styleProfiles.go
View file

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

62
scripts/fund-wallet.sh
View file

@ -1,62 +0,0 @@
#!/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
View file

@ -1,83 +0,0 @@
#!/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) ==="

17
sql/queries/AnalyticsCache.sql
View file

@ -1,17 +0,0 @@
-- 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();

11
sql/queries/AnalyticsCoverage.sql
View file

@ -1,11 +0,0 @@
-- 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();

17
sql/queries/AnalyticsPosts.sql
View file

@ -1,17 +0,0 @@
-- 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;

3
sql/queries/DeleteStylePosts.sql
View file

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

3
sql/queries/GetStylePostByID.sql
View file

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

5
sql/queries/GetStylePosts.sql
View file

@ -1,5 +0,0 @@
-- 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;

3
sql/queries/GetStylePostsByIDs.sql
View file

@ -1,3 +0,0 @@
-- 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[]);

3
sql/queries/PruneStylePosts.sql
View file

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

5
sql/queries/UpsertStylePosts.sql
View file

@ -1,5 +0,0 @@
-- 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();

2
sql/schema/001_transactions.sql
View file

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

4
sql/schema/002_user_style_profiles.sql
View file

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

2
sql/schema/003_add_cache_version.sql
View file

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

17
sql/schema/004_add_style_posts.sql
View file

@ -1,17 +0,0 @@
-- +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;

23
sql/schema/005_add_analytics_cache.sql
View file

@ -1,23 +0,0 @@
-- +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;

22
sql/schema/006_add_analytics_posts.sql
View file

@ -1,22 +0,0 @@
-- +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;

12
sql/schema/007_add_analytics_coverage.sql
View file

@ -1,12 +0,0 @@
-- +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;

291
testClient/main.go
View file

@ -1,22 +1,16 @@
package main package main
import ( import (
"bytes"
"context" "context"
"encoding/json"
"fmt" "fmt"
"io" "io"
"log" "log"
"math/big" "os"
"net/http" "net/http"
"net/http/httputil" "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" "github.com/joho/godotenv"
x402 "github.com/x402-foundation/x402/go/v2" x402 "github.com/x402-foundation/x402/go/v2"
@ -25,15 +19,6 @@ import (
evmsigners "github.com/x402-foundation/x402/go/v2/signers/evm" 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 { type LoggingRoundTripper struct {
Proxied http.RoundTripper Proxied http.RoundTripper
} }
@ -46,6 +31,7 @@ func (lrt *LoggingRoundTripper) RoundTrip(req *http.Request) (*http.Response, er
req.Body = rc req.Body = rc
} }
// Dump request
reqDump, err := httputil.DumpRequestOut(req, true) reqDump, err := httputil.DumpRequestOut(req, true)
if err != nil { if err != nil {
log.Printf("Error dumping request: %v", err) log.Printf("Error dumping request: %v", err)
@ -53,11 +39,13 @@ func (lrt *LoggingRoundTripper) RoundTrip(req *http.Request) (*http.Response, er
fmt.Printf("\n--- REQUEST ---\n%s\n----------------\n", reqDump) fmt.Printf("\n--- REQUEST ---\n%s\n----------------\n", reqDump)
} }
// Perform request
resp, err := lrt.Proxied.RoundTrip(req) resp, err := lrt.Proxied.RoundTrip(req)
if err != nil { if err != nil {
return nil, err return nil, err
} }
// Dump response
respDump, err := httputil.DumpResponse(resp, true) respDump, err := httputil.DumpResponse(resp, true)
if err != nil { if err != nil {
log.Printf("Error dumping response: %v", err) log.Printf("Error dumping response: %v", err)
@ -68,206 +56,13 @@ func (lrt *LoggingRoundTripper) RoundTrip(req *http.Request) (*http.Response, er
return resp, nil 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() { func main() {
godotenv.Load() godotenv.Load()
privateKey := os.Getenv("PRIVATE_KEY") var (
serverURL := getEnv("TEST_SERVER_URL", "http://localhost:8080") privateKey = os.Getenv("PRIVATE_KEY")
faucetURL := getEnv("FUNDING_FAUCET_URL", "") serverURL = "https://api.naco.li/v1"
rpcURL := getEnv("RPC_URL", BaseSepoliaRPC) ctx = context.Background()
)
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) signer, err := evmsigners.NewClientSignerFromPrivateKey(privateKey)
if err != nil { if err != nil {
@ -275,13 +70,12 @@ func main() {
} }
x402Client := x402.Newx402Client(). x402Client := x402.Newx402Client().
Register("eip155:84532", evm.NewExactEvmScheme(signer, nil)) Register("eip155:*", evm.NewExactEvmScheme(signer, nil))
debugClient := &http.Client{ debugClient := &http.Client{
Transport: &LoggingRoundTripper{ Transport: &LoggingRoundTripper{
Proxied: http.DefaultTransport, Proxied: http.DefaultTransport,
}, },
Timeout: 60 * time.Second,
} }
httpClient := x402http.WrapHTTPClientWithPayment( httpClient := x402http.WrapHTTPClientWithPayment(
@ -289,44 +83,39 @@ func main() {
x402http.Newx402HTTPClient(x402Client), x402http.Newx402HTTPClient(x402Client),
) )
log.Printf("Starting REST API endpoint tests against %s...", serverURL) log.Println("Starting REST API tests...")
testCases := []struct { log.Println("\nCalling getFeed (free tool) via HTTP client...")
name string bodyGetFeed := []byte(`{"uri": "https://bsky.app/profile/did:plc:z72i7hdynmk6r22z27h6tvur/feed/whats-hot"}`)
method string resp, err := httpClient.Post(fmt.Sprintf("%s/getFeed", serverURL), "application/json", bytes.NewReader(bodyGetFeed))
endpoint string if err != nil {
body string log.Println("getFeed HTTP request failed:", err)
}{ } else {
{"getFeed", "POST", "getFeed", `{"uri": "https://bsky.app/profile/did:plc:z72i7hdynmk6r22z27h6tvur/feed/whats-hot"}`}, defer resp.Body.Close()
{"getActor", "POST", "getActor", `{"author": "did:plc:z72i7hdynmk6r22z27h6tvur"}`}, body, _ := io.ReadAll(resp.Body)
{"searchActors", "POST", "searchActors", `{"query": "test", "limit": 5}`}, log.Printf("getFeed Response Status: %s\n", resp.Status)
{"getPost", "POST", "getPost", `{"actor":"sharkgamiart.bsky.social","rkey":"3m47y27c67c2b"}`}, log.Printf("getFeed Response Body: %s\n", string(body))
{"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}`},
} }
passed := 0 log.Println("\nCalling getPost (gated paid tool) via HTTP client...")
failed := 0 bodyGetPost := []byte(`{"actor":"sharkgamiart.bsky.social","rkey":"3m47y27c67c2b"}`)
for _, tc := range testCases { req, err := http.NewRequestWithContext(ctx, "POST", fmt.Sprintf("%s/getPost", serverURL), bytes.NewReader(bodyGetPost))
fmt.Printf("\n=== Testing %s ===\n", tc.name) if err != nil {
err := callEndpoint(httpClient, tc.method, tc.endpoint, tc.body) log.Fatal("Failed to create getPost request:", err)
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")
log.Printf("\n=== Test Summary ===") respPost, err := httpClient.Do(req)
log.Printf("Passed: %d", passed) if err != nil {
log.Printf("Failed: %d", failed) log.Println("getPost HTTP request failed:", err)
} else {
if failed > 0 { defer respPost.Body.Close()
os.Exit(1) body, _ := io.ReadAll(respPost.Body)
log.Printf("getPost Response Status: %s\n", respPost.Status)
log.Printf("getPost Response Body: %s\n", string(body))
} }
} }