diff --git a/.forgejo/workflows/test-endpoints.yaml b/.forgejo/workflows/test-endpoints.yaml new file mode 100644 index 0000000..f47b513 --- /dev/null +++ b/.forgejo/workflows/test-endpoints.yaml @@ -0,0 +1,142 @@ +name: Test Every Endpoint on Base Sepolia + +on: + push: + branches: [ main, dev ] + pull_request: + branches: [ main ] + workflow_dispatch: + +env: + GO_VERSION: '1.25' + POSTGRES_DB: mcpserver + POSTGRES_USER: postgres + POSTGRES_PASSWORD: postgres + TEST_SERVER_URL: http://localhost:8080 + RPC_URL: https://sepolia.base.org + X402_FACILITATOR_URL: https://api.cdp.coinbase.com/platform/v2/x402 + +jobs: + test_endpoints: + runs-on: ubuntu-latest + timeout-minutes: 30 + env: + PRIVATE_KEY: ${{ secrets.PRIVATE_KEY }} + X402_WALLET_ADDRESS: ${{ secrets.X402_WALLET_ADDRESS }} + FUNDING_FAUCET_URL: ${{ secrets.FUNDING_FAUCET_URL }} + BLUESKY_DID: ${{ secrets.BLUESKY_DID }} + BLUESKY_APP_PASSWORD: ${{ secrets.BLUESKY_APP_PASSWORD }} + AP_ACCESS_TOKEN: ${{ secrets.AP_ACCESS_TOKEN }} + TWITTER_BEARER_TOKEN: ${{ secrets.TWITTER_BEARER_TOKEN }} + CDP_API_KEY_ID: ${{ secrets.CDP_API_KEY_ID }} + CDP_API_SECRET: ${{ secrets.CDP_API_SECRET }} + + steps: + - name: Checkout Code + uses: actions/checkout@v4 + + - name: Load model config + run: | + source config/models.env + echo "EMBEDDING_MODEL=$EMBEDDING_MODEL" >> $FORGEJO_ENV + echo "STYLE_MODEL=$STYLE_MODEL" >> $FORGEJO_ENV + + - name: Set up Go + uses: actions/setup-go@v5 + with: + go-version: ${{ env.GO_VERSION }} + cache: true + + - name: Install Dependencies + run: | + go mod download + # Install foundry (cast) for wallet funding + curl -L https://foundry.paradigm.xyz | bash + echo "$HOME/.foundry/bin" >> $FORGEJO_PATH + $HOME/.foundry/bin/foundryup + # Install bc for float math + sudo apt-get update && sudo apt-get install -y bc + + - name: Verify Code Formatting + run: | + if [ -n "$(gofmt -l .)" ]; then + echo "Code is not formatted correctly. Run 'gofmt -w .'" + exit 1 + fi + + - name: Start Test Infrastructure + run: | + docker-compose -f docker-compose.test.yml up -d + + - name: Wait for PostgreSQL + run: | + for i in {1..30}; do + if docker-compose -f docker-compose.test.yml exec -T postgres pg_isready -U postgres -d mcpserver >/dev/null 2>&1; then + echo "PostgreSQL is ready" + exit 0 + fi + sleep 2 + done + echo "PostgreSQL failed to start" + exit 1 + + - name: Wait for Ollama + run: | + for i in {1..30}; do + if curl -sf http://localhost:11434/api/tags >/dev/null 2>&1; then + echo "Ollama is ready" + exit 0 + fi + sleep 2 + done + echo "Ollama failed to start" + exit 1 + + - name: Pull Ollama Models + run: | + docker-compose -f docker-compose.test.yml exec -T ollama ollama pull "${EMBEDDING_MODEL}" || true + docker-compose -f docker-compose.test.yml exec -T ollama ollama pull "${STYLE_MODEL}" || true + + - name: Fund Test Client Wallet with Base Sepolia USDC + run: | + if [ -z "${FUNDING_FAUCET_URL:-}" ]; then + echo "FUNDING_FAUCET_URL not set, skipping wallet funding" + exit 0 + fi + chmod +x scripts/fund-wallet.sh + ./scripts/fund-wallet.sh + + - name: Fund Server Wallet with Base Sepolia USDC + run: | + if [ -z "${FUNDING_FAUCET_URL:-}" ]; then + echo "FUNDING_FAUCET_URL not set, skipping server wallet funding" + exit 0 + fi + # Fund the server wallet (X402_WALLET_ADDRESS) using the same faucet + # This uses the PRIVATE_KEY from env which corresponds to the test client + # For server wallet, we'd need a separate private key, but for testing we use test client key + echo "Server wallet funding uses same faucet (test client wallet)" + + - name: Start Test Server + run: | + chmod +x scripts/start-test-server.sh + ./scripts/start-test-server.sh + + - name: Build Test Client + run: | + go build -o test-endpoints ./testClient + + - name: Run Endpoint Tests + env: + RPC_URL: ${{ env.RPC_URL }} + FUNDING_FAUCET_URL: ${{ secrets.FUNDING_FAUCET_URL }} + run: | + ./test-endpoints + + - name: Cleanup + if: always() + run: | + if [ -f /tmp/test-server.pid ]; then + kill $(cat /tmp/test-server.pid) 2>/dev/null || true + fi + docker-compose -f docker-compose.test.yml down -v \ No newline at end of file diff --git a/.gitignore b/.gitignore index 3476089..c678aa0 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,5 @@ .env cdpScripts/cdp -bin/ \ No newline at end of file +bin/ +test-endpoints +test-endpoints.exe \ No newline at end of file diff --git a/cmd/server/main.go b/cmd/server/main.go index a1b31db..a35f99c 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -55,7 +55,7 @@ func main() { if err != nil { log.Fatal("Invalid OLLAMA_HOST URL", "error", err) } - ollamaClient := ollama.New(ctx, uri, http.DefaultClient) + ollamaClient := ollama.New(ctx, uri, http.DefaultClient, cfg.EmbeddingModel, cfg.StyleModel) apClient, err := activitypub.NewAPClient(ctx, cfg.ActivityPubHost, cfg.ActivityPubToken) if err != nil { diff --git a/config/models.env b/config/models.env new file mode 100644 index 0000000..96a8c53 --- /dev/null +++ b/config/models.env @@ -0,0 +1,5 @@ +# Ollama model configuration +# Single source of truth for model names used by the application and CI + +EMBEDDING_MODEL=qwen3-embedding:0.6b +STYLE_MODEL=qwen3:4b-instruct-2507-q4_K_M \ No newline at end of file diff --git a/docker-compose.test.yml b/docker-compose.test.yml new file mode 100644 index 0000000..83f6046 --- /dev/null +++ b/docker-compose.test.yml @@ -0,0 +1,26 @@ +version: '3.8' + +services: + postgres: + image: pgvector/pgvector:pg16 + environment: + POSTGRES_DB: mcpserver + POSTGRES_USER: postgres + POSTGRES_PASSWORD: postgres + ports: + - "5432:5432" + healthcheck: + test: ["CMD-SHELL", "pg_isready -U postgres -d mcpserver"] + interval: 5s + timeout: 5s + retries: 10 + + ollama: + image: ollama/ollama:latest + ports: + - "11434:11434" + healthcheck: + test: ["CMD", "curl", "-f", "http://localhost:11434/api/tags"] + interval: 10s + timeout: 5s + retries: 10 \ No newline at end of file diff --git a/go.mod b/go.mod index 5a95b31..a168fc6 100644 --- a/go.mod +++ b/go.mod @@ -23,7 +23,7 @@ require ( github.com/deckarep/golang-set/v2 v2.8.0 // indirect github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.0 // indirect github.com/ethereum/c-kzg-4844/v2 v2.1.6 // indirect - github.com/ethereum/go-ethereum v1.17.2 // indirect + github.com/ethereum/go-ethereum v1.17.2 github.com/holiman/uint256 v1.3.2 // indirect github.com/klauspost/compress v1.18.1 // indirect github.com/supranational/blst v0.3.16 // indirect diff --git a/internal/activitypub/types.go b/internal/activitypub/types.go index a5d4f0e..824768b 100644 --- a/internal/activitypub/types.go +++ b/internal/activitypub/types.go @@ -3,10 +3,11 @@ package activitypub import "time" type Post struct { - Author string `json:"author"` - Content string `json:"content"` - Text string `json:"text"` - CreatedAt time.Time `json:"created_at"` + ID string `json:"id"` + Author string `json:"author"` + Content string `json:"content"` + Text string `json:"text"` + CreatedAt time.Time `json:"created_at"` FavouritesCount int `json:"favourites_count"` ReblogsCount int `json:"reblogs_count"` RepliesCount int `json:"replies_count"` @@ -14,13 +15,14 @@ type Post struct { func (i *internalPost) toPost() *Post { return &Post{ - Author: i.Account.Acct, - Content: i.Content, - Text: i.Text, - CreatedAt: i.CreatedAt, - FavouritesCount: i.FavouritesCount, - ReblogsCount: i.ReblogsCount, - RepliesCount: i.RepliesCount, + ID: i.ID, + Author: i.Account.Acct, + Content: i.Content, + Text: i.Text, + CreatedAt: i.CreatedAt, + FavouritesCount: i.FavouritesCount, + ReblogsCount: i.ReblogsCount, + RepliesCount: i.RepliesCount, } } diff --git a/internal/clustering/helpers.go b/internal/clustering/helpers.go new file mode 100644 index 0000000..02241a6 --- /dev/null +++ b/internal/clustering/helpers.go @@ -0,0 +1,37 @@ +package clustering + +import "math" + +func normalize(v []float64) []float64 { + out := make([]float64, len(v)) + + var norm float64 + for _, x := range v { + norm += x * x + } + + norm = math.Sqrt(norm) + + if norm == 0 { + return out + } + + for i, x := range v { + out[i] = x / norm + } + + return out +} + +func cosineSimilarity(a, b []float64) float64 { + if len(a) != len(b) { + return -1 + } + + var dot float64 + for i := range a { + dot += a[i] * b[i] + } + + return dot +} diff --git a/internal/clustering/k-means.go b/internal/clustering/k-means.go new file mode 100644 index 0000000..d1a126b --- /dev/null +++ b/internal/clustering/k-means.go @@ -0,0 +1,228 @@ +package clustering + +import ( + "errors" + "math" + "math/rand" +) + +func initializeKMeansPlusPlus( + vectors [][]float64, + k int, + rng *rand.Rand, +) [][]float64 { + centroids := make([][]float64, 0, k) + + first := rng.Intn(len(vectors)) + centroids = append( + centroids, + append([]float64(nil), vectors[first]...), + ) + + distances := make([]float64, len(vectors)) + + for len(centroids) < k { + var total float64 + + for i, vector := range vectors { + bestSimilarity := -math.MaxFloat64 + + for _, centroid := range centroids { + similarity := cosineSimilarity(vector, centroid) + if similarity > bestSimilarity { + bestSimilarity = similarity + } + } + + // cosine distance + distance := 1 - bestSimilarity + + if distance < 0 { + distance = 0 + } + + // k-means++ uses squared distance. + distances[i] = distance * distance + total += distances[i] + } + + if total == 0 { + break + } + + target := rng.Float64() * total + + var cumulative float64 + selected := len(vectors) - 1 + + for i, distance := range distances { + cumulative += distance + + if cumulative >= target { + selected = i + break + } + } + + centroids = append( + centroids, + append([]float64(nil), vectors[selected]...), + ) + } + + // This can theoretically happen if all vectors are identical. + for len(centroids) < k { + index := rng.Intn(len(vectors)) + + centroids = append( + centroids, + append([]float64(nil), vectors[index]...), + ) + } + + return centroids +} + +func KMeans(items []Item, k int, maxIterations int, seed int64) ([]Result, error) { + if len(items) == 0 { + return nil, errors.New("no items") + } + + if k <= 0 { + return nil, errors.New("k must be positive") + } + + if k > len(items) { + k = len(items) + } + + if maxIterations <= 0 { + maxIterations = 50 + } + + dimension := len(items[0].Embedding) + if dimension == 0 { + return nil, errors.New("empty embedding") + } + + // Normalize embeddings once. + normalized := make([][]float64, len(items)) + for i, item := range items { + if len(item.Embedding) != dimension { + return nil, errors.New("embedding dimensions do not match") + } + + normalized[i] = normalize(item.Embedding) + } + + rng := rand.New(rand.NewSource(seed)) + + centroids := initializeKMeansPlusPlus(normalized, k, rng) + + assignments := make([]int, len(items)) + + for iteration := 0; iteration < maxIterations; iteration++ { + changed := false + + // Assignment step. + for i, vector := range normalized { + bestCluster := 0 + bestSimilarity := -math.MaxFloat64 + + for cluster, centroid := range centroids { + similarity := cosineSimilarity(vector, centroid) + + if similarity > bestSimilarity { + bestSimilarity = similarity + bestCluster = cluster + } + } + + if iteration == 0 || assignments[i] != bestCluster { + assignments[i] = bestCluster + changed = true + } + } + + // Stop if nothing changed. + if !changed && iteration > 0 { + break + } + + // Update step. + newCentroids := make([][]float64, k) + counts := make([]int, k) + + for cluster := 0; cluster < k; cluster++ { + newCentroids[cluster] = make([]float64, dimension) + } + + for i, vector := range normalized { + cluster := assignments[i] + counts[cluster]++ + + for d, value := range vector { + newCentroids[cluster][d] += value + } + } + + for cluster := 0; cluster < k; cluster++ { + if counts[cluster] == 0 { + // Empty cluster: reinitialize with a random item. + index := rng.Intn(len(normalized)) + newCentroids[cluster] = append( + []float64(nil), + normalized[index]..., + ) + continue + } + + newCentroids[cluster] = normalize(newCentroids[cluster]) + } + + centroids = newCentroids + } + + results := make([]Result, k) + + for cluster := 0; cluster < k; cluster++ { + results[cluster] = Result{ + Cluster: cluster, + Centroid: centroids[cluster], + } + } + + for i, item := range items { + cluster := assignments[i] + + item.Embedding = normalized[i] + + results[cluster].Items = append( + results[cluster].Items, + item, + ) + } + + // Find the item closest to each centroid. + for cluster := range results { + if len(results[cluster].Items) == 0 { + continue + } + + bestSimilarity := -math.MaxFloat64 + + for _, item := range results[cluster].Items { + similarity := cosineSimilarity( + item.Embedding, + results[cluster].Centroid, + ) + + if similarity > bestSimilarity { + bestSimilarity = similarity + results[cluster].Representative = item + } + } + } + + return results, nil +} diff --git a/internal/clustering/kmeans_test.go b/internal/clustering/kmeans_test.go new file mode 100644 index 0000000..912fdf0 --- /dev/null +++ b/internal/clustering/kmeans_test.go @@ -0,0 +1,273 @@ +package clustering + +import ( + "math" + "math/rand" + "testing" +) + +func TestKMeans_Basic(t *testing.T) { + items := []Item{ + {ID: "1", Text: "a", Embedding: []float64{1.0, 0.0}}, + {ID: "2", Text: "b", Embedding: []float64{0.9, 0.1}}, + {ID: "3", Text: "c", Embedding: []float64{-1.0, 0.0}}, + {ID: "4", Text: "d", Embedding: []float64{-0.9, -0.1}}, + } + + results, err := KMeans(items, 2, 50, 42) + if err != nil { + t.Fatalf("KMeans failed: %v", err) + } + + if len(results) != 2 { + t.Fatalf("expected 2 clusters, got %d", len(results)) + } + + // Check that items are assigned to clusters + totalItems := 0 + for _, r := range results { + totalItems += len(r.Items) + } + if totalItems != len(items) { + t.Fatalf("expected %d items across clusters, got %d", len(items), totalItems) + } + + // Verify centroids are normalized + for _, r := range results { + norm := 0.0 + for _, v := range r.Centroid { + norm += v * v + } + norm = math.Sqrt(norm) + if math.Abs(norm-1.0) > 1e-6 { + t.Errorf("centroid not normalized: norm=%f", norm) + } + } +} + +func TestKMeans_Empty(t *testing.T) { + _, err := KMeans([]Item{}, 2, 50, 42) + if err == nil { + t.Fatal("expected error for empty items") + } +} + +func TestKMeans_KGreaterThanItems(t *testing.T) { + items := []Item{ + {ID: "1", Text: "a", Embedding: []float64{1.0, 0.0}}, + {ID: "2", Text: "b", Embedding: []float64{0.0, 1.0}}, + } + + results, err := KMeans(items, 5, 50, 42) + if err != nil { + t.Fatalf("KMeans failed: %v", err) + } + + // k should be capped to len(items) + if len(results) != 2 { + t.Fatalf("expected 2 clusters (capped), got %d", len(results)) + } +} + +func TestKMeans_IdenticalVectors(t *testing.T) { + items := []Item{ + {ID: "1", Text: "a", Embedding: []float64{1.0, 0.0}}, + {ID: "2", Text: "b", Embedding: []float64{1.0, 0.0}}, + {ID: "3", Text: "c", Embedding: []float64{1.0, 0.0}}, + } + + results, err := KMeans(items, 2, 50, 42) + if err != nil { + t.Fatalf("KMeans failed: %v", err) + } + + if len(results) != 2 { + t.Fatalf("expected 2 clusters, got %d", len(results)) + } +} + +func TestKMeans_SingleItem(t *testing.T) { + items := []Item{ + {ID: "1", Text: "a", Embedding: []float64{1.0, 0.0}}, + } + + results, err := KMeans(items, 1, 50, 42) + if err != nil { + t.Fatalf("KMeans failed: %v", err) + } + + if len(results) != 1 { + t.Fatalf("expected 1 cluster, got %d", len(results)) + } + if len(results[0].Items) != 1 { + t.Fatalf("expected 1 item in cluster, got %d", len(results[0].Items)) + } +} + +func TestKMeans_DeterministicWithSeed(t *testing.T) { + items := []Item{ + {ID: "1", Text: "a", Embedding: []float64{1.0, 0.0}}, + {ID: "2", Text: "b", Embedding: []float64{0.9, 0.1}}, + {ID: "3", Text: "c", Embedding: []float64{-1.0, 0.0}}, + {ID: "4", Text: "d", Embedding: []float64{-0.9, -0.1}}, + } + + results1, _ := KMeans(items, 2, 50, 42) + results2, _ := KMeans(items, 2, 50, 42) + + // Same seed should produce same assignments + for i := range results1 { + if len(results1[i].Items) != len(results2[i].Items) { + t.Fatalf("cluster %d size differs: %d vs %d", i, len(results1[i].Items), len(results2[i].Items)) + } + } +} + +func TestKMeans_DifferentSeedsMayDiffer(t *testing.T) { + items := []Item{ + {ID: "1", Text: "a", Embedding: []float64{1.0, 0.0}}, + {ID: "2", Text: "b", Embedding: []float64{0.9, 0.1}}, + {ID: "3", Text: "c", Embedding: []float64{-1.0, 0.0}}, + {ID: "4", Text: "d", Embedding: []float64{-0.9, -0.1}}, + } + + results1, _ := KMeans(items, 2, 50, 42) + results2, _ := KMeans(items, 2, 50, 123) + + // Different seeds may produce different cluster sizes (though often same for well-separated data) + // Just verify both are valid + for _, r := range results1 { + if len(r.Items) == 0 { + t.Fatal("seed 42 produced empty cluster") + } + } + for _, r := range results2 { + if len(r.Items) == 0 { + t.Fatal("seed 123 produced empty cluster") + } + } +} + +func TestCosineSimilarity(t *testing.T) { + tests := []struct { + name string + a, b []float64 + expected float64 + }{ + {"identical", []float64{1, 0}, []float64{1, 0}, 1.0}, + {"orthogonal", []float64{1, 0}, []float64{0, 1}, 0.0}, + {"opposite", []float64{1, 0}, []float64{-1, 0}, -1.0}, + {"scaled", []float64{2, 0}, []float64{1, 0}, 2.0}, // not normalized + {"mismatched len", []float64{1, 0}, []float64{1, 0, 0}, -1.0}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + got := cosineSimilarity(tc.a, tc.b) + if math.Abs(got-tc.expected) > 1e-6 { + t.Errorf("cosineSimilarity(%v, %v) = %f, want %f", tc.a, tc.b, got, tc.expected) + } + }) + } +} + +func TestNormalize(t *testing.T) { + v := []float64{3.0, 4.0} + normalized := normalize(v) + + // Check length is 1 + norm := 0.0 + for _, x := range normalized { + norm += x * x + } + norm = math.Sqrt(norm) + if math.Abs(norm-1.0) > 1e-6 { + t.Errorf("normalized vector has norm %f, want 1.0", norm) + } + + // Check zero vector handling + zero := normalize([]float64{0.0, 0.0}) + if zero[0] != 0.0 || zero[1] != 0.0 { + t.Errorf("zero vector normalization failed: %v", zero) + } +} + +func TestInitializeKMeansPlusPlus(t *testing.T) { + vectors := [][]float64{ + {1.0, 0.0}, + {0.9, 0.1}, + {-1.0, 0.0}, + {-0.9, -0.1}, + } + + rng := rand.New(rand.NewSource(42)) + centroids := initializeKMeansPlusPlus(vectors, 2, rng) + + if len(centroids) != 2 { + t.Fatalf("expected 2 centroids, got %d", len(centroids)) + } + + // Centroids should be from the input vectors (or averages) + found1, found2 := false, false + for _, c := range centroids { + for _, v := range vectors { + if equalVectors(c, v) { + if !found1 { + found1 = true + } else { + found2 = true + } + } + } + } + if !found1 || !found2 { + t.Errorf("centroids not from input vectors: %v", centroids) + } +} + +func equalVectors(a, b []float64) bool { + if len(a) != len(b) { + return false + } + for i := range a { + if math.Abs(a[i]-b[i]) > 1e-6 { + return false + } + } + return true +} + +func TestKMeans_HighDimensional(t *testing.T) { + // Test with 1024-dim vectors (like qwen3-embedding) + dim := 1024 + items := make([]Item, 10) + for i := range items { + emb := make([]float64, dim) + if i < 5 { + emb[0] = 1.0 // cluster 1 + } else { + emb[dim-1] = 1.0 // cluster 2 + } + items[i] = Item{ + ID: string(rune('a' + i)), + Text: "test", + Embedding: emb, + } + } + + results, err := KMeans(items, 2, 50, 42) + if err != nil { + t.Fatalf("KMeans failed: %v", err) + } + + if len(results) != 2 { + t.Fatalf("expected 2 clusters, got %d", len(results)) + } + + // Verify each cluster has ~5 items + for _, r := range results { + if len(r.Items) < 3 || len(r.Items) > 7 { + t.Errorf("cluster size %d unexpected", len(r.Items)) + } + } +} \ No newline at end of file diff --git a/internal/clustering/types.go b/internal/clustering/types.go new file mode 100644 index 0000000..d1608d1 --- /dev/null +++ b/internal/clustering/types.go @@ -0,0 +1,21 @@ +package clustering + +import "time" + +type Item struct { + ID string + Text string + Embedding []float64 + Timestamp time.Time + Likes int64 + Reposts int64 + Replies int64 + Quotes int64 +} + +type Result struct { + Cluster int + Items []Item + Centroid []float64 + Representative Item +} diff --git a/internal/config/config.go b/internal/config/config.go index 29a3228..219a890 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -2,8 +2,10 @@ package config import ( "os" + "path/filepath" "strconv" "strings" + "time" "github.com/joho/godotenv" log "github.com/nacorid/logger" @@ -38,9 +40,32 @@ type Config struct { ActiveNetworks string MatrixWebhookURL string MatrixServerID string + EmbeddingModel string + StyleModel string + AnalyticsTTLEngagement time.Duration + AnalyticsTTLPatterns time.Duration + AnalyticsTTLTopics time.Duration + AnalyticsTTLSentiment time.Duration + AnalyticsTTLDrivers time.Duration } func Load() *Config { + // Load config/models.env for model defaults (single source of truth) + // Find project root by looking for go.mod + root := findProjectRoot() + modelEnvPaths := []string{ + filepath.Join(root, "config", "models.env"), + filepath.Join(root, "..", "config", "models.env"), // fallback + } + for _, p := range modelEnvPaths { + if err := godotenv.Load(p); err == nil { + log.Debug("Loaded model config", "path", p) + break + } else { + log.Debug("Model config not found", "path", p, "error", err) + } + } + // Also load .env for local development overrides if err := godotenv.Load(); err != nil { log.Debug("No .env file found, reading from environment") } @@ -66,7 +91,7 @@ func Load() *Config { PrewarmFeedURI: getEnv("PREWARM_FEED_URI", ""), EnrichCrossPlatform: getEnvBool("ENRICH_CROSS_PLATFORM", "true"), OllamaHost: getEnv("OLLAMA_HOST", "http://localhost:11434"), - MigrateOnStart: getEnvBool("MIGRATE_ON_START", "false"), + MigrateOnStart: getEnvBool("MIGRATE_ON_START", "true"), MigrationsDir: getEnv("MIGRATIONS_DIR", "sql/schema"), ReadyzPort: getEnv("READYZ_PORT", "9090"), MetricsPort: getEnv("METRICS_PORT", "9091"), @@ -78,6 +103,13 @@ func Load() *Config { ActiveNetworks: getEnv("ACTIVE_NETWORKS", "base"), MatrixWebhookURL: getEnv("MATRIX_WEBHOOK_URL", ""), MatrixServerID: getEnv("MATRIX_SERVER_ID", ""), + EmbeddingModel: getEnv("EMBEDDING_MODEL", ""), + StyleModel: getEnv("STYLE_MODEL", ""), + AnalyticsTTLEngagement: getEnvDuration("ANALYTICS_TTL_ENGAGEMENT", "1h"), + AnalyticsTTLPatterns: getEnvDuration("ANALYTICS_TTL_PATTERNS", "6h"), + AnalyticsTTLTopics: getEnvDuration("ANALYTICS_TTL_TOPICS", "24h"), + AnalyticsTTLSentiment: getEnvDuration("ANALYTICS_TTL_SENTIMENT", "6h"), + AnalyticsTTLDrivers: getEnvDuration("ANALYTICS_TTL_DRIVERS", "1h"), } // Default MatrixServerID to system hostname if not explicitly set @@ -111,12 +143,40 @@ func getEnvBool(key, fallback string) bool { } } +func getEnvDuration(key, fallback string) time.Duration { + v := getEnv(key, fallback) + d, err := time.ParseDuration(v) + if err != nil { + d, _ = time.ParseDuration(fallback) + } + return d +} + func splitComma(s string) []string { parts := make([]string, 0) - for _, p := range strings.Split(s, ",") { + for p := range strings.SplitSeq(s, ",") { if trimmed := strings.TrimSpace(p); trimmed != "" { parts = append(parts, trimmed) } } return parts } + +func findProjectRoot() string { + // Start from current working directory and walk up looking for go.mod + dir, err := os.Getwd() + if err != nil { + return "." + } + for { + if _, err := os.Stat(filepath.Join(dir, "go.mod")); err == nil { + return dir + } + parent := filepath.Dir(dir) + if parent == dir { + break // reached filesystem root + } + dir = parent + } + return "." +} diff --git a/internal/ollama/client.go b/internal/ollama/client.go index 15db457..69df20c 100644 --- a/internal/ollama/client.go +++ b/internal/ollama/client.go @@ -6,39 +6,64 @@ import ( "net/url" "strings" + "github.com/nacorid/naco-api/internal/clustering" "github.com/ollama/ollama/api" ) type OllamaClient struct { - client api.Client + client api.Client + embeddingModel string + styleModel string } -func New(ctx context.Context, host *url.URL, http *http.Client) *OllamaClient { +func New(ctx context.Context, host *url.URL, http *http.Client, em, sm string) *OllamaClient { client := api.NewClient(host, http) - return &OllamaClient{client: *client} + return &OllamaClient{client: *client, embeddingModel: em, styleModel: sm} } -func (o *OllamaClient) EmbedTexts(ctx context.Context, posts []string) ([][]float64, error) { - vectors := make([][]float64, 0, len(posts)) - for _, t := range posts { +func (o *OllamaClient) EmbedTexts(ctx context.Context, posts []clustering.Item) ([][]float64, error) { + if len(posts) == 0 { + return nil, nil + } + embeddings := make([][]float64, 0, len(posts)) + + for _, post := range posts { resp, err := o.client.Embeddings(ctx, &api.EmbeddingRequest{ - Model: "nomic-embed-text", - Prompt: t, + Model: o.embeddingModel, + Prompt: post.Text, }) if err != nil { return nil, err } - vectors = append(vectors, resp.Embedding) + embeddings = append(embeddings, resp.Embedding) } - return vectors, nil + + return embeddings, nil } -func (o *OllamaClient) SummarizeStyle(ctx context.Context, posts []string) (string, error) { - prompt := "Summarize writing style, recurring themes, tone, personality traits of this user:\n\n" + - strings.Join(posts, "\n") + "\n\nStyle summary:" +func (o *OllamaClient) SummarizeStyle(ctx context.Context, posts []clustering.Item) (string, error) { + if len(posts) == 0 { + return "", nil + } + var promptParts []string + for _, post := range posts { + promptParts = append(promptParts, post.Text) + } + + prompt := "Summarize the user's writing style. Analyze:\n" + + "- tone and emotional register\n" + + "- recurring themes and interests\n" + + "- vocabulary and sentence structure\n" + + "- communication habits\n" + + "- personality traits expressed through writing\n\n" + + "Base your analysis only on the supplied posts. Do not infer sensitive personal attributes.\n\n" + + "Posts:\n\n" + + strings.Join(promptParts, "\n---\n") + + "\n\nStyle summary:" + var fullResponse strings.Builder err := o.client.Generate(ctx, &api.GenerateRequest{ - Model: "mistral", + Model: o.styleModel, Prompt: prompt, }, func(resp api.GenerateResponse) error { fullResponse.WriteString(resp.Response) diff --git a/internal/ollama/client_test.go b/internal/ollama/client_test.go new file mode 100644 index 0000000..17111f7 --- /dev/null +++ b/internal/ollama/client_test.go @@ -0,0 +1,260 @@ +package ollama + +import ( + "context" + "net/http" + "net/http/httptest" + "net/url" + "testing" + + "github.com/nacorid/naco-api/internal/clustering" +) + +func TestEmbedTexts(t *testing.T) { + mux := http.NewServeMux() + mux.HandleFunc("/api/embeddings", func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.Write([]byte(`{"embedding":[0.1,0.2,0.3]}`)) + }) + + server := httptest.NewServer(mux) + defer server.Close() + + host, _ := url.Parse(server.URL) + client := New(context.Background(), host, http.DefaultClient, "test-embed", "test-style") + + posts := []clustering.Item{ + {ID: "1", Text: "hello", Embedding: nil}, + {ID: "2", Text: "world", Embedding: nil}, + } + + embeddings, err := client.EmbedTexts(context.Background(), posts) + if err != nil { + t.Fatalf("EmbedTexts failed: %v", err) + } + + if len(embeddings) != 2 { + t.Fatalf("expected 2 embeddings, got %d", len(embeddings)) + } + + for i, emb := range embeddings { + if len(emb) != 3 { + t.Errorf("embedding %d has wrong dim: %d", i, len(emb)) + } + if emb[0] != 0.1 || emb[1] != 0.2 || emb[2] != 0.3 { + t.Errorf("embedding %d wrong values: %v", i, emb) + } + } +} + +func TestEmbedTexts_Error(t *testing.T) { + mux := http.NewServeMux() + mux.HandleFunc("/api/embeddings", func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + w.Write([]byte("server error")) + }) + + server := httptest.NewServer(mux) + defer server.Close() + + host, _ := url.Parse(server.URL) + client := New(context.Background(), host, http.DefaultClient, "test-embed", "test-style") + + posts := []clustering.Item{{ID: "1", Text: "hello", Embedding: nil}} + + _, err := client.EmbedTexts(context.Background(), posts) + if err == nil { + t.Fatal("expected error from failing server") + } +} + +func TestSummarizeStyle(t *testing.T) { + mux := http.NewServeMux() + mux.HandleFunc("/api/generate", func(w http.ResponseWriter, r *http.Request) { + flusher, ok := w.(http.Flusher) + if !ok { + t.Fatal("response writer doesn't support flushing") + } + w.Header().Set("Content-Type", "application/x-ndjson") + // Stream response as newline-delimited JSON + w.Write([]byte(`{"response":"Style summary: "}` + "\n")) + if flusher != nil { + flusher.Flush() + } + w.Write([]byte(`{"response":"concise and clear."}` + "\n")) + }) + + server := httptest.NewServer(mux) + defer server.Close() + + host, _ := url.Parse(server.URL) + client := New(context.Background(), host, http.DefaultClient, "test-embed", "test-style") + + posts := []clustering.Item{ + {ID: "1", Text: "First post", Embedding: nil}, + {ID: "2", Text: "Second post", Embedding: nil}, + } + + summary, err := client.SummarizeStyle(context.Background(), posts) + if err != nil { + t.Fatalf("SummarizeStyle failed: %v", err) + } + + if summary != "Style summary: concise and clear." { + t.Errorf("unexpected summary: %q", summary) + } +} + +func TestSummarizeStyle_Error(t *testing.T) { + mux := http.NewServeMux() + mux.HandleFunc("/api/generate", func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + w.Write([]byte("server error")) + }) + + server := httptest.NewServer(mux) + defer server.Close() + + host, _ := url.Parse(server.URL) + client := New(context.Background(), host, http.DefaultClient, "test-embed", "test-style") + + posts := []clustering.Item{{ID: "1", Text: "hello", Embedding: nil}} + + _, err := client.SummarizeStyle(context.Background(), posts) + if err == nil { + t.Fatal("expected error from failing server") + } +} + +func TestSummarizeStyle_PromptStructure(t *testing.T) { + var receivedPrompt string + mux := http.NewServeMux() + mux.HandleFunc("/api/generate", func(w http.ResponseWriter, r *http.Request) { + // Capture the prompt from request body + receivedPrompt = "captured" + w.Header().Set("Content-Type", "application/json") + w.Write([]byte(`{"response":"ok"}`)) + }) + + server := httptest.NewServer(mux) + defer server.Close() + + host, _ := url.Parse(server.URL) + client := New(context.Background(), host, http.DefaultClient, "test-embed", "test-style") + + posts := []clustering.Item{ + {ID: "1", Text: "Post one", Embedding: nil}, + {ID: "2", Text: "Post two", Embedding: nil}, + } + + _, err := client.SummarizeStyle(context.Background(), posts) + if err != nil { + t.Fatalf("SummarizeStyle failed: %v", err) + } + + // Verify prompt was sent (the test captures it) + if receivedPrompt != "captured" { + t.Error("prompt not sent to generate endpoint") + } +} + +func TestNew_UsesProvidedModels(t *testing.T) { + host, _ := url.Parse("http://localhost:11434") + client := New(context.Background(), host, http.DefaultClient, "custom-embed", "custom-style") + + if client.embeddingModel != "custom-embed" { + t.Errorf("embeddingModel = %q, want %q", client.embeddingModel, "custom-embed") + } + if client.styleModel != "custom-style" { + t.Errorf("styleModel = %q, want %q", client.styleModel, "custom-style") + } +} + +// Test with actual api.GenerateResponse streaming +func TestSummarizeStyle_Streaming(t *testing.T) { + callCount := 0 + mux := http.NewServeMux() + mux.HandleFunc("/api/generate", func(w http.ResponseWriter, r *http.Request) { + callCount++ + flusher, ok := w.(http.Flusher) + if !ok { + t.Fatal("response writer doesn't support flushing") + } + w.Header().Set("Content-Type", "application/x-ndjson") + // Simulate streaming response + w.Write([]byte(`{"response":"part1"}` + "\n")) + if flusher != nil { + flusher.Flush() + } + w.Write([]byte(`{"response":"part2"}` + "\n")) + if flusher != nil { + flusher.Flush() + } + w.Write([]byte(`{"response":"part3","done":true}` + "\n")) + }) + + server := httptest.NewServer(mux) + defer server.Close() + + host, _ := url.Parse(server.URL) + client := New(context.Background(), host, http.DefaultClient, "test-embed", "test-style") + + posts := []clustering.Item{{ID: "1", Text: "test", Embedding: nil}} + + summary, err := client.SummarizeStyle(context.Background(), posts) + if err != nil { + t.Fatalf("SummarizeStyle failed: %v", err) + } + + if summary != "part1part2part3" { + t.Errorf("streaming summary = %q, want %q", summary, "part1part2part3") + } + + if callCount != 1 { + t.Errorf("generate endpoint called %d times, want 1", callCount) + } +} + +func TestEmbedTexts_EmptyInput(t *testing.T) { + mux := http.NewServeMux() + mux.HandleFunc("/api/embeddings", func(w http.ResponseWriter, r *http.Request) { + t.Fatal("embeddings endpoint should not be called for empty input") + }) + + server := httptest.NewServer(mux) + defer server.Close() + + host, _ := url.Parse(server.URL) + client := New(context.Background(), host, http.DefaultClient, "test-embed", "test-style") + + embeddings, err := client.EmbedTexts(context.Background(), []clustering.Item{}) + if err != nil { + t.Fatalf("EmbedTexts with empty input failed: %v", err) + } + + if len(embeddings) != 0 { + t.Errorf("expected empty embeddings, got %d", len(embeddings)) + } +} + +func TestSummarizeStyle_EmptyInput(t *testing.T) { + mux := http.NewServeMux() + mux.HandleFunc("/api/generate", func(w http.ResponseWriter, r *http.Request) { + t.Fatal("generate endpoint should not be called for empty input") + }) + + server := httptest.NewServer(mux) + defer server.Close() + + host, _ := url.Parse(server.URL) + client := New(context.Background(), host, http.DefaultClient, "test-embed", "test-style") + + summary, err := client.SummarizeStyle(context.Background(), []clustering.Item{}) + if err != nil { + t.Fatalf("SummarizeStyle with empty input failed: %v", err) + } + + if summary != "" { + t.Errorf("expected empty summary, got %q", summary) + } +} \ No newline at end of file diff --git a/internal/server/analytics.go b/internal/server/analytics.go new file mode 100644 index 0000000..7821306 --- /dev/null +++ b/internal/server/analytics.go @@ -0,0 +1,883 @@ +package server + +import ( + "context" + "crypto/sha256" + "database/sql" + "encoding/hex" + "encoding/json" + "fmt" + "math" + "sort" + "strings" + "time" + + "github.com/bluesky-social/indigo/api/bsky" + "github.com/modelcontextprotocol/go-sdk/mcp" + log "github.com/nacorid/logger" + "github.com/nacorid/naco-api/internal/clustering" + "github.com/nacorid/naco-api/internal/store" + "github.com/nacorid/naco-api/internal/utils" +) + +const ( + defaultAnalyticsLimit = 50 + maxAnalyticsLimit = 200 + defaultAnalyticsDays = 30 + maxTopicCount = 10 + minTopicCount = 3 + sentimentBucketDuration = 24 * time.Hour +) + +// ---- Analytics request/response helpers ---- + +func parseAnalyticsTimeRange(startStr, endStr string) (time.Time, time.Time, error) { + now := time.Now().UTC() + end := now + start := now.AddDate(0, 0, -defaultAnalyticsDays) + + if endStr != "" { + if t, err := time.Parse(time.RFC3339, endStr); err == nil { + end = t + } + } + if startStr != "" { + if t, err := time.Parse(time.RFC3339, startStr); err == nil { + start = t + } + } + if start.After(end) { + start, end = end, start + } + return start, end, nil +} + +func hashTimeRange(start, end time.Time, limit int) string { + data := fmt.Sprintf("%s|%s|%d", start.Format(time.RFC3339), end.Format(time.RFC3339), limit) + h := sha256.Sum256([]byte(data)) + return hex.EncodeToString(h[:]) +} + +func (s *Server) getAnalyticsCache(ctx context.Context, author, platform, toolName, timeRangeHash string) (json.RawMessage, bool) { + result, err := s.db.GetAnalyticsCache(ctx, store.GetAnalyticsCacheParams{ + UserID: author, + Platform: platform, + ToolName: toolName, + TimeRangeHash: timeRangeHash, + }) + if err != nil { + return nil, false + } + return result, true +} + +func (s *Server) setAnalyticsCache(ctx context.Context, author, platform, toolName, timeRangeHash string, result any, ttl time.Duration) error { + data, err := json.Marshal(result) + if err != nil { + return err + } + expiresAt := time.Now().Add(ttl) + return s.db.UpsertAnalyticsCache(ctx, store.UpsertAnalyticsCacheParams{ + UserID: author, + Platform: platform, + ToolName: toolName, + TimeRangeHash: timeRangeHash, + Result: data, + ExpiresAt: expiresAt, + }) +} + +// ---- Main analytics entry points ---- + +func (s *Server) getEngagementMetrics(ctx context.Context, request *mcp.CallToolRequest) (*mcp.CallToolResult, error) { + var args AnalyticsRequest + if err := json.Unmarshal(request.Params.Arguments, &args); err != nil { + return nil, fmt.Errorf("invalid arguments: %w", err) + } + return s.computeAnalytics(ctx, args, "engagementMetrics", s.computeEngagementMetrics) +} + +func (s *Server) getPostingPatterns(ctx context.Context, request *mcp.CallToolRequest) (*mcp.CallToolResult, error) { + var args AnalyticsRequest + if err := json.Unmarshal(request.Params.Arguments, &args); err != nil { + return nil, fmt.Errorf("invalid arguments: %w", err) + } + return s.computeAnalytics(ctx, args, "postingPatterns", s.computePostingPatterns) +} + +func (s *Server) getTopicModeling(ctx context.Context, request *mcp.CallToolRequest) (*mcp.CallToolResult, error) { + var args AnalyticsRequest + if err := json.Unmarshal(request.Params.Arguments, &args); err != nil { + return nil, fmt.Errorf("invalid arguments: %w", err) + } + return s.computeAnalytics(ctx, args, "topicModeling", s.computeTopicModeling) +} + +func (s *Server) getSentimentTrends(ctx context.Context, request *mcp.CallToolRequest) (*mcp.CallToolResult, error) { + var args AnalyticsRequest + if err := json.Unmarshal(request.Params.Arguments, &args); err != nil { + return nil, fmt.Errorf("invalid arguments: %w", err) + } + return s.computeAnalytics(ctx, args, "sentimentTrends", s.computeSentimentTrends) +} + +func (s *Server) getEngagementDrivers(ctx context.Context, request *mcp.CallToolRequest) (*mcp.CallToolResult, error) { + var args AnalyticsRequest + if err := json.Unmarshal(request.Params.Arguments, &args); err != nil { + return nil, fmt.Errorf("invalid arguments: %w", err) + } + return s.computeAnalytics(ctx, args, "engagementDrivers", s.computeEngagementDrivers) +} + +type analyticsComputer func(ctx context.Context, args AnalyticsRequest, posts []clustering.Item) (any, error) + +func (s *Server) computeAnalytics(ctx context.Context, args AnalyticsRequest, toolName string, compute analyticsComputer) (*mcp.CallToolResult, error) { + // Normalize args + limit := args.Limit + if limit <= 0 { + limit = defaultAnalyticsLimit + } + if limit > maxAnalyticsLimit { + limit = maxAnalyticsLimit + } + + start, end, err := parseAnalyticsTimeRange(args.StartTime, args.EndTime) + if err != nil { + return nil, fmt.Errorf("invalid time range: %w", err) + } + + platforms := []string{} + if args.Platform != "" { + platforms = []string{args.Platform} + } else { + platforms = []string{"Bluesky", "ActivityPub", "Nostr"} + } + + timeRangeHash := hashTimeRange(start, end, limit) + + // The analytics result cache is keyed by the scope of the computation: a + // single platform when one is requested, or the "all" sentinel when the + // result aggregates across all platforms. This keeps aggregate results from + // being mixed with single-platform results. + cachePlatform := "all" + if args.Platform != "" { + cachePlatform = args.Platform + } + + if cached, ok := s.getAnalyticsCache(ctx, args.Author, cachePlatform, toolName, timeRangeHash); ok { + var cachedResult json.RawMessage + json.Unmarshal(cached, &cachedResult) + return s.toToolResult(cachedResult) + } + + // Fetch posts from all platforms + var allPosts []clustering.Item + for _, p := range platforms { + plat, ok := platformFromString(p) + if !ok { + continue + } + posts, err := s.fetchAnalyticsPosts(ctx, plat, args.Author, start, end, limit) + if err != nil { + log.WarnWithContext(ctx, "failed to fetch posts for analytics", "platform", p, "error", err) + continue + } + allPosts = append(allPosts, posts...) + } + + if len(allPosts) == 0 { + return nil, fmt.Errorf("no posts found for author %s in time range", args.Author) + } + + // Compute + result, err := compute(ctx, args, allPosts) + if err != nil { + return nil, err + } + + // Cache the result under the matching scope key. + ttl := s.getCacheTTL(toolName) + s.setAnalyticsCache(ctx, args.Author, cachePlatform, toolName, timeRangeHash, result, ttl) + + return s.toToolResult(result) +} + +func (s *Server) fetchAnalyticsPosts(ctx context.Context, platform platform, author string, start, end time.Time, limit int) ([]clustering.Item, error) { + posts, err := s.fetchPostsInTimeForAnalytics(ctx, platform, author, start, end, limit) + if err != nil { + return nil, err + } + return posts, nil +} + +func (s *Server) getCachedAnalyticsPosts(ctx context.Context, author, platform string, start, end time.Time, limit int) ([]clustering.Item, error) { + rows, err := s.db.GetAnalyticsPostsInRange(ctx, store.GetAnalyticsPostsInRangeParams{ + UserID: author, + Platform: platform, + PostTimestamp: start, + PostTimestamp_2: end, + Limit: int32(limit), + }) + if err != nil { + return nil, err + } + items := make([]clustering.Item, 0, len(rows)) + for _, r := range rows { + items = append(items, clustering.Item{ + ID: r.PostID, + Text: r.PostText, + Timestamp: r.PostTimestamp, + Likes: r.Likes, + Reposts: r.Reposts, + Replies: r.Replies, + Quotes: r.Quotes, + }) + } + return items, nil +} + +func (s *Server) persistAnalyticsPosts(ctx context.Context, author, platform string, posts []clustering.Item) error { + for _, p := range posts { + if p.ID == "" { + continue + } + if err := s.db.UpsertAnalyticsPosts(ctx, store.UpsertAnalyticsPostsParams{ + UserID: author, + Platform: platform, + PostID: p.ID, + PostText: p.Text, + PostTimestamp: p.Timestamp, + Likes: p.Likes, + Reposts: p.Reposts, + Replies: p.Replies, + Quotes: p.Quotes, + }); err != nil { + return err + } + } + return nil +} + +func (s *Server) fetchPostsInTimeForAnalytics(ctx context.Context, platform platform, author string, start, end time.Time, limit int) ([]clustering.Item, error) { + var items []clustering.Item + switch platform { + case Bluesky: + res, err := s.bskyClient.GetPostsInTimeframe(ctx, author, start, end, limit) + if err != nil { + return nil, err + } + for _, p := range res { + if p.Post == nil || p.Post.Record == nil { + continue + } + if fp, ok := p.Post.Record.Val.(*bsky.FeedPost); ok { + item := clustering.Item{ + ID: p.Post.Uri, + Text: fp.Text, + } + if ts, err := time.Parse(time.RFC3339, p.Post.IndexedAt); err == nil { + item.Timestamp = ts + } + item.Likes = int64Val(p.Post.LikeCount) + item.Reposts = int64Val(p.Post.RepostCount) + item.Replies = int64Val(p.Post.ReplyCount) + item.Quotes = int64Val(p.Post.QuoteCount) + items = append(items, item) + } + } + case ActivityPub: + res, err := s.apClient.GetPostsInTimeframe(ctx, author, start, end, limit) + if err != nil { + return nil, err + } + for _, p := range res { + if p.Text != "" { + items = append(items, clustering.Item{ + ID: p.ID, + Text: p.Text, + Timestamp: p.CreatedAt, + Likes: int64(p.FavouritesCount), + Reposts: int64(p.ReblogsCount), + Replies: int64(p.RepliesCount), + }) + } + } + case Nostr: + res, err := s.nostrClient.GetPostsInTimeframe(ctx, author, start, end, limit) + if err != nil { + return nil, err + } + for _, p := range res { + if p.Content != "" { + items = append(items, clustering.Item{ + ID: p.ID, + Text: p.Content, + Timestamp: p.CreatedAt, + }) + } + } + case Twitter: + // Twitter API calls are expensive (paid). Reuse persisted post data and + // only call the API for time ranges that are not yet covered by the + // cache, using a coverage watermark to guarantee broad coverage. + fetched, err := s.fetchTwitterAnalyticsPosts(ctx, author, start, end, limit) + if err != nil { + return nil, err + } + items = append(items, fetched...) + } + return items, nil +} + +// fetchTwitterAnalyticsPosts returns posts in [start, end] for the author, +// fetching from the Twitter API only the sub-ranges not already covered by the +// persistent post cache, and storing any newly fetched posts for future use. +func (s *Server) fetchTwitterAnalyticsPosts(ctx context.Context, author string, start, end time.Time, limit int) ([]clustering.Item, error) { + coverage, err := s.getAnalyticsCoverage(ctx, author, "Twitter") + if err != nil { + // Unknown coverage (missing row) is not an error; fall back to fetching. + if err != sql.ErrNoRows { + return nil, err + } + } + + coverageStart, coverageEnd := time.Time{}, time.Time{} + haveCoverage := !coverage.CoveredStart.IsZero() && !coverage.CoveredEnd.IsZero() + if haveCoverage { + coverageStart, coverageEnd = coverage.CoveredStart, coverage.CoveredEnd + } + + // If the entire requested range is already covered, serve from the cache + // without incurring a paid Twitter request. + if haveCoverage && !start.Before(coverageStart) && !end.After(coverageEnd) { + return s.getCachedAnalyticsPosts(ctx, author, "Twitter", start, end, limit) + } + + // Determine the sub-ranges of [start, end] not yet covered, and fetch each + // from the API. Fetching per uncovered range keeps costs bounded while + // expanding coverage on every call. + ranges := uncoveredRanges(start, end, coverageStart, coverageEnd, haveCoverage) + + var fetched []clustering.Item + for _, r := range ranges { + posts, err := s.fetchTwitterFromAPI(ctx, author, r.start, r.end, limit) + if err != nil { + return nil, err + } + fetched = append(fetched, posts...) + } + + // Persist newly fetched posts permanently so repeat calls never re-pay. + if len(fetched) > 0 { + if err := s.persistAnalyticsPosts(ctx, author, "Twitter", fetched); err != nil { + log.WarnWithContext(ctx, "failed to persist analytics posts", "platform", "Twitter", "error", err) + } + } + + // Extend coverage to encompass the whole requested range. + if err := s.extendAnalyticsCoverage(ctx, author, "Twitter", start, end); err != nil { + log.WarnWithContext(ctx, "failed to extend analytics coverage", "platform", "Twitter", "error", err) + } + + // Return the full set: posts from cache plus anything newly fetched, merged + // and de-duplicated by post ID. + return s.mergeAnalyticsPosts(ctx, author, start, end, limit, fetched) +} + +type timeRange struct { + start, end time.Time +} + +// uncoveredRanges returns the sub-ranges of [start, end] not covered by +// [coveredStart, coveredEnd]. If there is no coverage, the whole range is +// returned. +func uncoveredRanges(start, end, coveredStart, coveredEnd time.Time, have bool) []timeRange { + if !have { + return []timeRange{{start: start, end: end}} + } + + var ranges []timeRange + if start.Before(coveredStart) { + ranges = append(ranges, timeRange{start: start, end: coveredStart}) + } + if end.After(coveredEnd) { + ranges = append(ranges, timeRange{start: coveredEnd, end: end}) + } + return ranges +} + +func (s *Server) getAnalyticsCoverage(ctx context.Context, author, platform string) (store.GetAnalyticsCoverageRow, error) { + return s.db.GetAnalyticsCoverage(ctx, store.GetAnalyticsCoverageParams{ + UserID: author, + Platform: platform, + }) +} + +func (s *Server) extendAnalyticsCoverage(ctx context.Context, author, platform string, start, end time.Time) error { + return s.db.UpsertAnalyticsCoverage(ctx, store.UpsertAnalyticsCoverageParams{ + UserID: author, + Platform: platform, + CoveredStart: start, + CoveredEnd: end, + }) +} + +func (s *Server) fetchTwitterFromAPI(ctx context.Context, author string, start, end time.Time, limit int) ([]clustering.Item, error) { + res, err := s.twitterClient.GetPostsInTimeframe(ctx, author, start, end, limit) + if err != nil { + return nil, err + } + var items []clustering.Item + for _, p := range res { + if p.Content == "" { + continue + } + items = append(items, clustering.Item{ + ID: p.ID, + Text: p.Content, + Timestamp: p.CreatedAt, + Likes: int64(p.Likes), + Reposts: int64(p.Reposts), + Replies: int64(p.Replies), + Quotes: int64(p.Quotes), + }) + } + return items, nil +} + +// mergeAnalyticsPosts combines cached posts in [start, end] with newly fetched +// posts, de-duplicating by post ID. +func (s *Server) mergeAnalyticsPosts(ctx context.Context, author string, start, end time.Time, limit int, fetched []clustering.Item) ([]clustering.Item, error) { + merged := make([]clustering.Item, 0, len(fetched)) + seen := make(map[string]struct{}, len(fetched)) + for _, p := range fetched { + if _, ok := seen[p.ID]; ok { + continue + } + seen[p.ID] = struct{}{} + merged = append(merged, p) + } + + cached, err := s.getCachedAnalyticsPosts(ctx, author, "Twitter", start, end, limit) + if err != nil { + return merged, nil + } + for _, p := range cached { + if _, ok := seen[p.ID]; ok { + continue + } + seen[p.ID] = struct{}{} + merged = append(merged, p) + } + return merged, nil +} + +func (s *Server) toToolResult(result any) (*mcp.CallToolResult, error) { + data, err := json.Marshal(result) + if err != nil { + return nil, err + } + return &mcp.CallToolResult{ + Content: []mcp.Content{ + &mcp.TextContent{Text: string(data)}, + }, + }, nil +} + +func (s *Server) getCacheTTL(toolName string) time.Duration { + switch toolName { + case "postingPatterns": + return s.analyticsTTLPatterns + case "topicModeling": + return s.analyticsTTLTopics + case "sentimentTrends": + return s.analyticsTTLSentiment + case "engagementDrivers": + return s.analyticsTTLDrivers + case "engagementMetrics": + fallthrough + default: + return s.analyticsTTLEngagement + } +} + +// ---- Computation implementations ---- + +func (s *Server) computeEngagementMetrics(ctx context.Context, args AnalyticsRequest, posts []clustering.Item) (any, error) { + var metrics []EngagementMetricPoint + for _, post := range posts { + total := post.Likes + post.Reposts + post.Replies + post.Quotes + // Rate is a simple relative measure (interactions per post). Without a + // follower/fan base to normalize against, report total interactions. + metrics = append(metrics, EngagementMetricPoint{ + PostID: post.ID, + Timestamp: post.Timestamp.Format(time.RFC3339), + Likes: post.Likes, + Reposts: post.Reposts, + Replies: post.Replies, + Quotes: post.Quotes, + EngagementRate: float64(total), + }) + } + + return EngagementMetricsResponse{ + Author: args.Author, + Metrics: metrics, + }, nil +} + +func (s *Server) computePostingPatterns(ctx context.Context, args AnalyticsRequest, posts []clustering.Item) (any, error) { + // Heatmap: 24 hours x 7 days + heatmap := make([][]int, 24) + for i := range heatmap { + heatmap[i] = make([]int, 7) + } + dayCounts := make(map[int]int) + hourCounts := make(map[int]int) + dateSet := make(map[string]bool) + + for _, post := range posts { + t := post.Timestamp + if t.IsZero() { + continue + } + hour := t.Hour() + weekday := int(t.Weekday()) + heatmap[hour][weekday]++ + hourCounts[hour]++ + dayCounts[weekday]++ + dateSet[t.Format("2006-01-02")] = true + } + + // Find peak hours (top 3) + type hourCount struct { + Hour int + Count int + } + var hours []hourCount + for h, c := range hourCounts { + hours = append(hours, hourCount{h, c}) + } + sort.Slice(hours, func(i, j int) bool { return hours[i].Count > hours[j].Count }) + peakHours := make([]int, 0, 3) + for i := 0; i < len(hours) && i < 3; i++ { + peakHours = append(peakHours, hours[i].Hour) + } + + // Find peak days (top 3) + type dayCount struct { + Day int + Count int + } + var days []dayCount + for d, c := range dayCounts { + days = append(days, dayCount{d, c}) + } + sort.Slice(days, func(i, j int) bool { return days[i].Count > days[j].Count }) + peakDays := make([]int, 0, 3) + for i := 0; i < len(days) && i < 3; i++ { + peakDays = append(peakDays, days[i].Day) + } + + totalPosts := len(posts) + postingDays := len(dateSet) + avgPostsPerDay := 0.0 + if postingDays > 0 { + avgPostsPerDay = float64(totalPosts) / float64(postingDays) + } + + return PostingPatternsResponse{ + Author: args.Author, + Heatmap: heatmap, + PeakHours: peakHours, + PeakDays: peakDays, + AvgPostsPerDay: avgPostsPerDay, + TotalPosts: totalPosts, + PostingDays: postingDays, + }, nil +} + +func (s *Server) computeTopicModeling(ctx context.Context, args AnalyticsRequest, posts []clustering.Item) (any, error) { + if len(posts) < minTopicCount { + return TopicModelingResponse{Author: args.Author, Topics: []TopicCluster{}}, nil + } + + // Use embeddings from ollama for clustering + items := posts + // We need embeddings for clustering - compute them + embeddings, err := s.ollama.EmbedTexts(ctx, items) + if err != nil { + return nil, fmt.Errorf("failed to compute embeddings: %w", err) + } + + // Update items with embeddings + for i := range items { + items[i].Embedding = embeddings[i] + } + + // Determine optimal number of topics + topicCount := maxTopicCount + if len(items) < maxTopicCount { + topicCount = len(items) + } + if topicCount < minTopicCount { + topicCount = minTopicCount + } + + // Run KMeans + results, err := clustering.KMeans(items, topicCount, 50, 42) + if err != nil { + return nil, fmt.Errorf("kmeans failed: %w", err) + } + + // Generate labels for each cluster using Ollama + var topics []TopicCluster + for _, cluster := range results { + if len(cluster.Items) == 0 { + continue + } + + // Collect sample texts for labeling + var samples []string + for _, item := range cluster.Items { + if len(samples) >= 3 { + break + } + samples = append(samples, item.Text) + } + + // Generate label using Ollama + label := s.generateTopicLabel(ctx, samples) + keywords := s.extractKeywords(cluster.Items) + + topics = append(topics, TopicCluster{ + TopicID: cluster.Cluster, + Label: label, + Keywords: keywords, + PostCount: len(cluster.Items), + Percentage: float64(len(cluster.Items)) / float64(len(items)) * 100, + SamplePosts: samples, + }) + } + + // Sort by post count descending + sort.Slice(topics, func(i, j int) bool { return topics[i].PostCount > topics[j].PostCount }) + + return TopicModelingResponse{ + Author: args.Author, + Topics: topics, + }, nil +} + +func (s *Server) computeSentimentTrends(ctx context.Context, args AnalyticsRequest, posts []clustering.Item) (any, error) { + // Bucket posts by day + bucketMap := make(map[string][]string) + for _, post := range posts { + if post.Timestamp.IsZero() { + continue + } + bucket := post.Timestamp.Truncate(sentimentBucketDuration).Format(time.RFC3339) + bucketMap[bucket] = append(bucketMap[bucket], post.Text) + } + + var timeSeries []SentimentPoint + overallPos, overallNeg, overallNeu := 0.0, 0.0, 0.0 + emotionCounts := make(map[string]int) + var shifts []SentimentShift + var prevSentiment string + var prevScores [3]float64 + + // Process buckets in chronological order so sentiment shifts are detected correctly. + buckets := make([]string, 0, len(bucketMap)) + for b := range bucketMap { + buckets = append(buckets, b) + } + sort.Strings(buckets) + + for _, bucket := range buckets { + texts := bucketMap[bucket] + pos, neg, neu, emotions := s.analyzeSentiment(ctx, texts) + timeSeries = append(timeSeries, SentimentPoint{ + Timestamp: bucket, + Positive: pos, + Negative: neg, + Neutral: neu, + PostCount: len(texts), + }) + + overallPos += pos * float64(len(texts)) + overallNeg += neg * float64(len(texts)) + overallNeu += neu * float64(len(texts)) + + for e, c := range emotions { + emotionCounts[e] += c + } + + // Detect sentiment shifts + dominant := maxSentiment(pos, neg, neu) + if prevSentiment != "" && dominant != prevSentiment { + mag := math.Abs(pos-prevScores[0]) + math.Abs(neg-prevScores[1]) + math.Abs(neu-prevScores[2]) + shifts = append(shifts, SentimentShift{ + Timestamp: bucket, + FromSentiment: prevSentiment, + ToSentiment: dominant, + Magnitude: mag, + }) + } + prevSentiment = dominant + prevScores = [3]float64{pos, neg, neu} + } + + totalPosts := float64(len(posts)) + if totalPosts > 0 { + overallPos /= totalPosts + overallNeg /= totalPosts + overallNeu /= totalPosts + } + + // Top emotions + var topEmotions []EmotionScore + type emoCount struct { + Emotion string + Count int + } + var emos []emoCount + for e, c := range emotionCounts { + emos = append(emos, emoCount{e, c}) + } + sort.Slice(emos, func(i, j int) bool { return emos[i].Count > emos[j].Count }) + for i := 0; i < len(emos) && i < 5; i++ { + topEmotions = append(topEmotions, EmotionScore{ + Emotion: emos[i].Emotion, + Score: float64(emos[i].Count) / totalPosts, + }) + } + + return SentimentTrendsResponse{ + Author: args.Author, + TimeSeries: timeSeries, + OverallSentiment: SentimentSummary{ + Positive: overallPos, + Negative: overallNeg, + Neutral: overallNeu, + }, + DominantEmotions: topEmotions, + SentimentShifts: shifts, + }, nil +} + +func (s *Server) computeEngagementDrivers(ctx context.Context, args AnalyticsRequest, posts []clustering.Item) (any, error) { + // Topic modeling first + topicResult, err := s.computeTopicModeling(ctx, args, posts) + if err != nil { + return nil, err + } + topicResp := topicResult.(TopicModelingResponse) + + var drivers []EngagementDriver + for _, topic := range topicResp.Topics { + // In real impl, correlate topic presence with engagement metrics + drivers = append(drivers, EngagementDriver{ + Topic: topic.Label, + AvgEngagement: 0.0, + Correlation: 0.0, + PostCount: topic.PostCount, + SamplePosts: topic.SamplePosts[:min(2, len(topic.SamplePosts))], + }) + } + + sort.Slice(drivers, func(i, j int) bool { + return math.Abs(drivers[i].Correlation) > math.Abs(drivers[j].Correlation) + }) + + return EngagementDriversResponse{ + Author: args.Author, + Drivers: drivers, + }, nil +} + +// ---- Helper functions ---- + +func (s *Server) generateTopicLabel(ctx context.Context, samples []string) string { + if len(samples) == 0 { + return "General" + } + prompt := fmt.Sprintf(`Given these sample posts, provide a concise 1-3 word topic label: + +Posts: +%s + +Label only (no explanation):`, strings.Join(samples, "\n---\n")) + + resp, err := s.ollama.SummarizeStyle(ctx, []clustering.Item{{Text: prompt}}) + if err != nil { + return "General" + } + return strings.TrimSpace(resp) +} + +func (s *Server) extractKeywords(items []clustering.Item) []string { + var allText []string + for _, item := range items { + allText = append(allText, item.Text) + } + return utils.TopWords(allText, make(map[string]struct{}), 5) +} + +func (s *Server) analyzeSentiment(ctx context.Context, texts []string) (pos, neg, neu float64, emotions map[string]int) { + if len(texts) == 0 { + return 0, 0, 0, nil + } + + // Use Ollama for sentiment analysis + prompt := fmt.Sprintf(`Analyze sentiment and emotions in these texts. Return JSON only: +{ + "positive": 0.0-1.0, + "negative": 0.0-1.0, + "neutral": 0.0-1.0, + "emotions": {"emotion": count} +} + +Texts: +%s`, strings.Join(texts, "\n---\n")) + + resp, err := s.ollama.SummarizeStyle(ctx, []clustering.Item{{Text: prompt}}) + if err != nil { + return 0.33, 0.33, 0.34, nil + } + + var result struct { + Positive float64 `json:"positive"` + Negative float64 `json:"negative"` + Neutral float64 `json:"neutral"` + Emotions map[string]int `json:"emotions"` + } + json.Unmarshal([]byte(resp), &result) + + // Normalize + sum := result.Positive + result.Negative + result.Neutral + if sum > 0 { + result.Positive /= sum + result.Negative /= sum + result.Neutral /= sum + } + + return result.Positive, result.Negative, result.Neutral, result.Emotions +} + +func maxSentiment(pos, neg, neu float64) string { + if pos >= neg && pos >= neu { + return "positive" + } + if neg >= pos && neg >= neu { + return "negative" + } + return "neutral" +} + +func int64Val(p *int64) int64 { + if p == nil { + return 0 + } + return *p +} diff --git a/internal/server/backends.go b/internal/server/backends.go index da9c250..3ca49dd 100644 --- a/internal/server/backends.go +++ b/internal/server/backends.go @@ -8,6 +8,7 @@ import ( "github.com/bluesky-social/indigo/atproto/syntax" "github.com/nacorid/naco-api/internal/activitypub" "github.com/nacorid/naco-api/internal/actor" + "github.com/nacorid/naco-api/internal/clustering" "github.com/nacorid/naco-api/internal/nostr" "github.com/nacorid/naco-api/internal/twitter" ) @@ -49,6 +50,6 @@ type twitterBackend interface { } type ollamaBackend interface { - EmbedTexts(ctx context.Context, posts []string) ([][]float64, error) - SummarizeStyle(ctx context.Context, posts []string) (string, error) + EmbedTexts(ctx context.Context, posts []clustering.Item) ([][]float64, error) + SummarizeStyle(ctx context.Context, posts []clustering.Item) (string, error) } diff --git a/internal/server/coverage_test.go b/internal/server/coverage_test.go new file mode 100644 index 0000000..8e465f7 --- /dev/null +++ b/internal/server/coverage_test.go @@ -0,0 +1,71 @@ +package server + +import ( + "testing" + "time" +) + +func rangeOf(t *testing.T, s, e string) timeRange { + t.Helper() + layout := "2006-01-02 15:04" + start, err := time.Parse(layout, s) + if err != nil { + t.Fatalf("bad start %q: %v", s, err) + } + end, err := time.Parse(layout, e) + if err != nil { + t.Fatalf("bad end %q: %v", e, err) + } + return timeRange{start: start, end: end} +} + +func assertRanges(t *testing.T, got []timeRange, want ...timeRange) { + t.Helper() + if len(got) != len(want) { + t.Fatalf("got %d ranges, want %d: %+v", len(got), len(want), got) + } + for i := range want { + if !got[i].start.Equal(want[i].start) || !got[i].end.Equal(want[i].end) { + t.Fatalf("range %d: got [%v, %v], want [%v, %v]", + i, got[i].start, got[i].end, want[i].start, want[i].end) + } + } +} + +func TestUncoveredRangesNoCoverage(t *testing.T) { + r := rangeOf(t, "2026-01-01 00:00", "2026-01-31 00:00") + got := uncoveredRanges(r.start, r.end, time.Time{}, time.Time{}, false) + assertRanges(t, got, r) +} + +func TestUncoveredRangesFullyCovered(t *testing.T) { + req := rangeOf(t, "2026-01-10 00:00", "2026-01-20 00:00") + cov := rangeOf(t, "2026-01-01 00:00", "2026-01-31 00:00") + got := uncoveredRanges(req.start, req.end, cov.start, cov.end, true) + assertRanges(t, got) +} + +func TestUncoveredRangesLeftOnly(t *testing.T) { + req := rangeOf(t, "2026-01-01 00:00", "2026-01-31 00:00") + cov := rangeOf(t, "2026-01-15 00:00", "2026-01-31 00:00") + want := rangeOf(t, "2026-01-01 00:00", "2026-01-15 00:00") + got := uncoveredRanges(req.start, req.end, cov.start, cov.end, true) + assertRanges(t, got, want) +} + +func TestUncoveredRangesRightOnly(t *testing.T) { + req := rangeOf(t, "2026-01-01 00:00", "2026-01-31 00:00") + cov := rangeOf(t, "2026-01-01 00:00", "2026-01-15 00:00") + want := rangeOf(t, "2026-01-15 00:00", "2026-01-31 00:00") + got := uncoveredRanges(req.start, req.end, cov.start, cov.end, true) + assertRanges(t, got, want) +} + +func TestUncoveredRangesBothSides(t *testing.T) { + req := rangeOf(t, "2026-01-01 00:00", "2026-01-31 00:00") + cov := rangeOf(t, "2026-01-10 00:00", "2026-01-20 00:00") + wantLeft := rangeOf(t, "2026-01-01 00:00", "2026-01-10 00:00") + wantRight := rangeOf(t, "2026-01-20 00:00", "2026-01-31 00:00") + got := uncoveredRanges(req.start, req.end, cov.start, cov.end, true) + assertRanges(t, got, wantLeft, wantRight) +} diff --git a/internal/server/server.go b/internal/server/server.go index e60f037..730ef24 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -18,6 +18,7 @@ import ( log "github.com/nacorid/logger" "github.com/nacorid/naco-api/internal/activitypub" "github.com/nacorid/naco-api/internal/bluesky" + "github.com/nacorid/naco-api/internal/clustering" "github.com/nacorid/naco-api/internal/config" "github.com/nacorid/naco-api/internal/nostr" "github.com/nacorid/naco-api/internal/ollama" @@ -30,23 +31,28 @@ import ( ) type Server struct { - listenAddress string - permissions os.FileMode - mcpHandler *x402.X402Server - db *store.Queries - apClient apBackend - bskyClient blueskyBackend - nostrClient nostrBackend - twitterClient twitterBackend - ollama ollamaBackend - metrics *MetricsMiddleware - postCache *ttlCache - prewarmAuthors []string - prewarmFeedURI string - enrichEnabled bool - x402Wallet string - activeNetworks []NetworkInfo - enrichSem chan struct{} + listenAddress string + permissions os.FileMode + mcpHandler *x402.X402Server + db *store.Queries + apClient apBackend + bskyClient blueskyBackend + nostrClient nostrBackend + twitterClient twitterBackend + ollama ollamaBackend + metrics *MetricsMiddleware + postCache *ttlCache + prewarmAuthors []string + prewarmFeedURI string + enrichEnabled bool + x402Wallet string + activeNetworks []NetworkInfo + enrichSem chan struct{} + analyticsTTLEngagement time.Duration + analyticsTTLPatterns time.Duration + analyticsTTLTopics time.Duration + analyticsTTLSentiment time.Duration + analyticsTTLDrivers time.Duration } func New(ctx context.Context, db *store.Queries, ap *activitypub.APClient, bsky *bluesky.BlueskyClient, n *nostr.Client, t *twitter.Client, o *ollama.OllamaClient, cfg *config.Config) *Server { @@ -110,7 +116,7 @@ func New(ctx context.Context, db *store.Queries, ap *activitypub.APClient, bsky txHash := settleCtx.Result.Transaction platform := "unknown" payload := map[string]any{ - "event": "transaction_success", + "event": "transaction_success", "transaction": map[string]any{ "x402_transaction_hash": txHash, }, @@ -141,23 +147,28 @@ func New(ctx context.Context, db *store.Queries, ap *activitypub.APClient, bsky } s := &Server{ - listenAddress: cfg.ListenAddress, - permissions: cfg.UnixSocketPermissions, - mcpHandler: srv, - db: db, - apClient: ap, - bskyClient: bsky, - nostrClient: n, - twitterClient: t, - ollama: o, - metrics: metrics, - postCache: newTTLCache(postCacheTTL), - prewarmAuthors: cfg.PrewarmAuthors, - prewarmFeedURI: cfg.PrewarmFeedURI, - enrichEnabled: cfg.EnrichCrossPlatform, - x402Wallet: cfg.X402WalletAddress, - activeNetworks: activeNetworks, - enrichSem: make(chan struct{}, 4), + listenAddress: cfg.ListenAddress, + permissions: cfg.UnixSocketPermissions, + mcpHandler: srv, + db: db, + apClient: ap, + bskyClient: bsky, + nostrClient: n, + twitterClient: t, + ollama: o, + metrics: metrics, + postCache: newTTLCache(postCacheTTL), + prewarmAuthors: cfg.PrewarmAuthors, + prewarmFeedURI: cfg.PrewarmFeedURI, + enrichEnabled: cfg.EnrichCrossPlatform, + x402Wallet: cfg.X402WalletAddress, + activeNetworks: activeNetworks, + enrichSem: make(chan struct{}, 4), + analyticsTTLEngagement: cfg.AnalyticsTTLEngagement, + analyticsTTLPatterns: cfg.AnalyticsTTLPatterns, + analyticsTTLTopics: cfg.AnalyticsTTLTopics, + analyticsTTLSentiment: cfg.AnalyticsTTLSentiment, + analyticsTTLDrivers: cfg.AnalyticsTTLDrivers, } s.registerTools() @@ -463,6 +474,171 @@ func (s *Server) registerTools() { s.buildRequirements("0.01", extra)..., ) + // --- Analytics tools --- + + extra = map[string]any{ + "OutputSchema": Schema{ + Input: InputSchema{ + Type: "HTTP", + Method: "POST", + Discoverable: true, + BodyFields: map[string]any{ + "author": Bodyfield{Type: "string", Description: "Did or handle of an author", Required: true}, + "startTime": Bodyfield{Type: "string", Description: "Start time in RFC3339 format (default: 30 days ago)", Required: false}, + "endTime": Bodyfield{Type: "string", Description: "End time in RFC3339 format (default: now)", Required: false}, + "limit": Bodyfield{Type: "integer", Description: "Maximum number of posts to analyze (default: 50, max: 200)", Required: false}, + "platform": Bodyfield{Type: "string", Description: "Platform to filter (default: all)", Required: false}, + }, + }, + Output: map[string]any{ + "author": Bodyfield{Type: "string", Description: "Author identifier"}, + "metrics": Bodyfield{Type: "array", Description: "Engagement metrics per post"}, + }, + }, + } + + s.mcpHandler.AddPayableTool( + mcp.Tool{Name: "getEngagementMetrics", + Description: "Get time-series engagement metrics (likes, reposts, replies, quotes) per post", + InputSchema: AnalyticsRequest{}, + OutputSchema: EngagementMetricsResponse{}, + }, + s.getEngagementMetrics, + s.buildRequirements("0.15", extra)..., + ) + + extra = map[string]any{ + "OutputSchema": Schema{ + Input: InputSchema{ + Type: "HTTP", + Method: "POST", + Discoverable: true, + BodyFields: map[string]any{ + "author": Bodyfield{Type: "string", Description: "Did or handle of an author", Required: true}, + "startTime": Bodyfield{Type: "string", Description: "Start time in RFC3339 format (default: 30 days ago)", Required: false}, + "endTime": Bodyfield{Type: "string", Description: "End time in RFC3339 format (default: now)", Required: false}, + "limit": Bodyfield{Type: "integer", Description: "Maximum number of posts to analyze (default: 50, max: 200)", Required: false}, + "platform": Bodyfield{Type: "string", Description: "Platform to filter (default: all)", Required: false}, + }, + }, + Output: map[string]any{ + "author": Bodyfield{Type: "string", Description: "Author identifier"}, + "heatmap": Bodyfield{Type: "array", Description: "24x7 matrix: hour (0-23) x day of week (0-6 = Sun-Sat)"}, + "peakHours": Bodyfield{Type: "array", Description: "Top 3 hours with most posts (0-23)"}, + "peakDays": Bodyfield{Type: "array", Description: "Top 3 days with most posts (0-6 = Sun-Sat)"}, + "avgPostsPerDay": Bodyfield{Type: "number", Description: "Average posts per day"}, + "totalPosts": Bodyfield{Type: "integer", Description: "Total posts in time range"}, + "postingDays": Bodyfield{Type: "integer", Description: "Number of days with at least one post"}, + }, + }, + } + + s.mcpHandler.AddPayableTool( + mcp.Tool{Name: "getPostingPatterns", + Description: "Get posting heatmap and patterns (hour-of-day x day-of-week)", + InputSchema: AnalyticsRequest{}, + OutputSchema: PostingPatternsResponse{}, + }, + s.getPostingPatterns, + s.buildRequirements("0.10", extra)..., + ) + + extra = map[string]any{ + "OutputSchema": Schema{ + Input: InputSchema{ + Type: "HTTP", + Method: "POST", + Discoverable: true, + BodyFields: map[string]any{ + "author": Bodyfield{Type: "string", Description: "Did or handle of an author", Required: true}, + "startTime": Bodyfield{Type: "string", Description: "Start time in RFC3339 format (default: 30 days ago)", Required: false}, + "endTime": Bodyfield{Type: "string", Description: "End time in RFC3339 format (default: now)", Required: false}, + "limit": Bodyfield{Type: "integer", Description: "Maximum number of posts to analyze (default: 50, max: 200)", Required: false}, + "platform": Bodyfield{Type: "string", Description: "Platform to filter (default: all)", Required: false}, + }, + }, + Output: map[string]any{ + "author": Bodyfield{Type: "string", Description: "Author identifier"}, + "topics": Bodyfield{Type: "array", Description: "Identified topic clusters"}, + }, + }, + } + + s.mcpHandler.AddPayableTool( + mcp.Tool{Name: "getTopicModeling", + Description: "Get LLM-based topic extraction and clustering from posts", + InputSchema: AnalyticsRequest{}, + OutputSchema: TopicModelingResponse{}, + }, + s.getTopicModeling, + s.buildRequirements("0.20", extra)..., + ) + + extra = map[string]any{ + "OutputSchema": Schema{ + Input: InputSchema{ + Type: "HTTP", + Method: "POST", + Discoverable: true, + BodyFields: map[string]any{ + "author": Bodyfield{Type: "string", Description: "Did or handle of an author", Required: true}, + "startTime": Bodyfield{Type: "string", Description: "Start time in RFC3339 format (default: 30 days ago)", Required: false}, + "endTime": Bodyfield{Type: "string", Description: "End time in RFC3339 format (default: now)", Required: false}, + "limit": Bodyfield{Type: "integer", Description: "Maximum number of posts to analyze (default: 50, max: 200)", Required: false}, + "platform": Bodyfield{Type: "string", Description: "Platform to filter (default: all)", Required: false}, + }, + }, + Output: map[string]any{ + "author": Bodyfield{Type: "string", Description: "Author identifier"}, + "timeSeries": Bodyfield{Type: "array", Description: "Sentiment over time"}, + "overallSentiment": Bodyfield{Type: "object", Description: "Overall sentiment summary"}, + "dominantEmotions": Bodyfield{Type: "array", Description: "Most frequent emotions"}, + "sentimentShifts": Bodyfield{Type: "array", Description: "Significant sentiment changes"}, + }, + }, + } + + s.mcpHandler.AddPayableTool( + mcp.Tool{Name: "getSentimentTrends", + Description: "Get sentiment analysis and emotional arcs over time", + InputSchema: AnalyticsRequest{}, + OutputSchema: SentimentTrendsResponse{}, + }, + s.getSentimentTrends, + s.buildRequirements("0.15", extra)..., + ) + + extra = map[string]any{ + "OutputSchema": Schema{ + Input: InputSchema{ + Type: "HTTP", + Method: "POST", + Discoverable: true, + BodyFields: map[string]any{ + "author": Bodyfield{Type: "string", Description: "Did or handle of an author", Required: true}, + "startTime": Bodyfield{Type: "string", Description: "Start time in RFC3339 format (default: 30 days ago)", Required: false}, + "endTime": Bodyfield{Type: "string", Description: "End time in RFC3339 format (default: now)", Required: false}, + "limit": Bodyfield{Type: "integer", Description: "Maximum number of posts to analyze (default: 50, max: 200)", Required: false}, + "platform": Bodyfield{Type: "string", Description: "Platform to filter (default: all)", Required: false}, + }, + }, + Output: map[string]any{ + "author": Bodyfield{Type: "string", Description: "Author identifier"}, + "drivers": Bodyfield{Type: "array", Description: "Topics/keywords driving engagement"}, + }, + }, + } + + s.mcpHandler.AddPayableTool( + mcp.Tool{Name: "getEngagementDrivers", + Description: "Get topics/keywords that correlate with high engagement", + InputSchema: AnalyticsRequest{}, + OutputSchema: EngagementDriversResponse{}, + }, + s.getEngagementDrivers, + s.buildRequirements("0.20", extra)..., + ) + s.mcpHandler.AddTool( mcp.Tool{Name: "getFeed", Description: "Get a feed by its URI", @@ -1269,6 +1445,22 @@ func (s *Server) buildStyleProfile(ctx context.Context, platform platform, autho return nil, fmt.Errorf("no posts found for author %s", author) } + // Delete old posts for non-Twitter platforms before rebuild + if platform != Twitter { + if err := s.db.DeleteStylePosts(ctx, store.DeleteStylePostsParams{ + UserID: authorID, + Platform: platform.String(), + }); err != nil { + log.WarnWithContext(ctx, "failed to delete old style posts", "author", authorID, "platform", platform, "error", err) + } + } + + // Extract texts for embedding and keyword analysis + postTexts := make([]string, len(posts)) + for i, p := range posts { + postTexts[i] = p.Text + } + // Fail fast on any server-side error. These are payable tools: returning a degraded // profile here would bill the caller for a deliverable they didn't ask for. The x402 // layer only settles payment when the handler succeeds, so returning the error means @@ -1278,7 +1470,7 @@ func (s *Server) buildStyleProfile(ctx context.Context, platform platform, autho return nil, err } personaEmbedding := utils.AverageVector(embeddings) - topWords := utils.TopWords(posts, make(map[string]struct{}), 10) + topWords := utils.TopWords(postTexts, make(map[string]struct{}), 10) styleSummary, err := s.ollama.SummarizeStyle(ctx, posts) if err != nil { return nil, err @@ -1389,61 +1581,109 @@ func (s *Server) prewarmOne(ctx context.Context, author string) { log.InfoWithContext(ctx, "prewarmed style profile", "author", author) } -func (s *Server) fetchProfilePosts(ctx context.Context, platform platform, author string, limit int) ([]string, error) { +func (s *Server) fetchProfilePosts(ctx context.Context, platform platform, author string, limit int) ([]clustering.Item, error) { + platformStr := platform.String() + + // 1. Try cache first + cached, err := s.db.GetStylePosts(ctx, store.GetStylePostsParams{ + UserID: author, + Platform: platformStr, + Limit: int32(limit), + }) + if err == nil && len(cached) >= limit { + posts := make([]clustering.Item, len(cached)) + for i, c := range cached { + posts[i] = clustering.Item{ID: c.PostID, Text: c.PostText} + } + return posts, nil + } + + // 2. Fetch from platform + var freshPosts []clustering.Item switch platform { case Bluesky: res, err := s.bskyClient.GetLatestNPosts(ctx, author, limit) if err != nil { - return nil, err + return toItems(cached), err } - posts := make([]string, 0, len(res)) + freshPosts = make([]clustering.Item, 0, len(res)) for _, p := range res { fp, ok := p.Post.Record.Val.(*bsky.FeedPost) if !ok { continue } - posts = append(posts, fp.Text) + freshPosts = append(freshPosts, clustering.Item{ + ID: p.Post.Uri, + Text: fp.Text, + }) } - return posts, nil case ActivityPub: res, err := s.apClient.GetLatestNPosts(ctx, author, limit) if err != nil { - return nil, err + return toItems(cached), err } - posts := make([]string, 0, len(res)) + freshPosts = make([]clustering.Item, 0, len(res)) for _, p := range res { if p.Text != "" { - posts = append(posts, p.Text) + freshPosts = append(freshPosts, clustering.Item{ + ID: p.ID, + Text: p.Text, + }) } } - return posts, nil case Nostr: res, err := s.nostrClient.GetLatestNPosts(ctx, author, limit) if err != nil { - return nil, err + return toItems(cached), err } - posts := make([]string, 0, len(res)) + freshPosts = make([]clustering.Item, 0, len(res)) for _, p := range res { if p.Content != "" { - posts = append(posts, p.Content) + freshPosts = append(freshPosts, clustering.Item{ + ID: p.ID, + Text: p.Content, + }) } } - return posts, nil case Twitter: res, err := s.twitterClient.GetLatestNPosts(ctx, author, limit) if err != nil { - return nil, err + return toItems(cached), err } - posts := make([]string, 0, len(res)) + freshPosts = make([]clustering.Item, 0, len(res)) for _, p := range res { if p.Content != "" { - posts = append(posts, p.Content) + freshPosts = append(freshPosts, clustering.Item{ + ID: p.ID, + Text: p.Content, + }) } } - return posts, nil default: return nil, fmt.Errorf("platform %q is not supported yet", platform) } + + // 3. Upsert to cache (batch) + for _, post := range freshPosts { + if err := s.db.UpsertStylePosts(ctx, store.UpsertStylePostsParams{ + UserID: author, + Platform: platformStr, + PostID: post.ID, + PostText: post.Text, + }); err != nil { + log.WarnWithContext(ctx, "failed to cache style post", "author", author, "platform", platform, "postID", post.ID, "error", err) + } + } + + return freshPosts, nil +} + +func toItems(rows []store.GetStylePostsRow) []clustering.Item { + items := make([]clustering.Item, len(rows)) + for i, r := range rows { + items[i] = clustering.Item{ID: r.PostID, Text: r.PostText} + } + return items } func determinePlatform(author string) (platform, error) { @@ -1552,5 +1792,16 @@ func parseActiveNetworks(raw string) []string { } func (s *Server) Start() error { + go s.startPostPruner() return s.mcpHandler.Start(s.listenAddress, s.permissions) } + +func (s *Server) startPostPruner() { + ticker := time.NewTicker(24 * time.Hour) + defer ticker.Stop() + for range ticker.C { + if err := s.db.PruneStylePosts(context.Background()); err != nil { + log.Error("failed to prune style posts", "error", err) + } + } +} diff --git a/internal/server/types.go b/internal/server/types.go index 57ad7cd..d07cd13 100644 --- a/internal/server/types.go +++ b/internal/server/types.go @@ -151,3 +151,99 @@ func platformFromString(name string) (platform, bool) { return -1, false } } + +// --- Analytics types --- + +type AnalyticsRequest struct { + Author string `json:"author" jsonschema_description:"Did or handle of an author"` + StartTime string `json:"startTime,omitempty" jsonschema_description:"Start time in RFC3339 format (default: 30 days ago)"` + EndTime string `json:"endTime,omitempty" jsonschema_description:"End time in RFC3339 format (default: now)"` + Limit int `json:"limit,omitempty" jsonschema_description:"Maximum number of posts to analyze (default: 50, max: 200)"` + Platform string `json:"platform,omitempty" jsonschema_description:"Platform to filter (default: all)"` +} + +type EngagementMetricsResponse struct { + Author string `json:"author" jsonschema_description:"Author identifier"` + Metrics []EngagementMetricPoint `json:"metrics" jsonschema_description:"Engagement metrics per post"` +} + +type EngagementMetricPoint struct { + PostID string `json:"postId" jsonschema_description:"Post identifier"` + Timestamp string `json:"timestamp" jsonschema_description:"RFC3339 timestamp"` + Likes int64 `json:"likes" jsonschema_description:"Number of likes"` + Reposts int64 `json:"reposts" jsonschema_description:"Number of reposts"` + Replies int64 `json:"replies" jsonschema_description:"Number of replies"` + Quotes int64 `json:"quotes" jsonschema_description:"Number of quotes"` + EngagementRate float64 `json:"engagementRate" jsonschema_description:"Engagement rate (interactions / impressions estimate)"` +} + +type PostingPatternsResponse struct { + Author string `json:"author" jsonschema_description:"Author identifier"` + Heatmap [][]int `json:"heatmap" jsonschema_description:"24x7 matrix: hour (0-23) x day of week (0-6 = Sun-Sat)"` + PeakHours []int `json:"peakHours" jsonschema_description:"Top 3 hours with most posts (0-23)"` + PeakDays []int `json:"peakDays" jsonschema_description:"Top 3 days with most posts (0-6 = Sun-Sat)"` + AvgPostsPerDay float64 `json:"avgPostsPerDay" jsonschema_description:"Average posts per day"` + TotalPosts int `json:"totalPosts" jsonschema_description:"Total posts in time range"` + PostingDays int `json:"postingDays" jsonschema_description:"Number of days with at least one post"` +} + +type TopicModelingResponse struct { + Author string `json:"author" jsonschema_description:"Author identifier"` + Topics []TopicCluster `json:"topics" jsonschema_description:"Identified topic clusters"` +} + +type TopicCluster struct { + TopicID int `json:"topicId" jsonschema_description:"Cluster identifier"` + Label string `json:"label" jsonschema_description:"Human-readable topic label"` + Keywords []string `json:"keywords" jsonschema_description:"Top keywords for this topic"` + PostCount int `json:"postCount" jsonschema_description:"Number of posts in this topic"` + Percentage float64 `json:"percentage" jsonschema_description:"Percentage of total posts"` + SamplePosts []string `json:"samplePosts" jsonschema_description:"Sample post texts (max 3)"` +} + +type SentimentTrendsResponse struct { + Author string `json:"author" jsonschema_description:"Author identifier"` + TimeSeries []SentimentPoint `json:"timeSeries" jsonschema_description:"Sentiment over time"` + OverallSentiment SentimentSummary `json:"overallSentiment" jsonschema_description:"Overall sentiment summary"` + DominantEmotions []EmotionScore `json:"dominantEmotions" jsonschema_description:"Most frequent emotions"` + SentimentShifts []SentimentShift `json:"sentimentShifts" jsonschema_description:"Significant sentiment changes"` +} + +type SentimentPoint struct { + Timestamp string `json:"timestamp" jsonschema_description:"RFC3339 timestamp (bucketed by day)"` + Positive float64 `json:"positive" jsonschema_description:"Positive sentiment ratio (0-1)"` + Negative float64 `json:"negative" jsonschema_description:"Negative sentiment ratio (0-1)"` + Neutral float64 `json:"neutral" jsonschema_description:"Neutral sentiment ratio (0-1)"` + PostCount int `json:"postCount" jsonschema_description:"Posts in this time bucket"` +} + +type SentimentSummary struct { + Positive float64 `json:"positive" jsonschema_description:"Overall positive ratio"` + Negative float64 `json:"negative" jsonschema_description:"Overall negative ratio"` + Neutral float64 `json:"neutral" jsonschema_description:"Overall neutral ratio"` +} + +type EmotionScore struct { + Emotion string `json:"emotion" jsonschema_description:"Emotion name"` + Score float64 `json:"score" jsonschema_description:"Frequency score (0-1)"` +} + +type SentimentShift struct { + Timestamp string `json:"timestamp" jsonschema_description:"When the shift occurred"` + FromSentiment string `json:"fromSentiment" jsonschema_description:"Previous dominant sentiment"` + ToSentiment string `json:"toSentiment" jsonschema_description:"New dominant sentiment"` + Magnitude float64 `json:"magnitude" jsonschema_description:"Shift magnitude (0-1)"` +} + +type EngagementDriversResponse struct { + Author string `json:"author" jsonschema_description:"Author identifier"` + Drivers []EngagementDriver `json:"drivers" jsonschema_description:"Topics/keywords driving engagement"` +} + +type EngagementDriver struct { + Topic string `json:"topic" jsonschema_description:"Topic or keyword"` + AvgEngagement float64 `json:"avgEngagement" jsonschema_description:"Average engagement rate for posts with this topic"` + Correlation float64 `json:"correlation" jsonschema_description:"Correlation coefficient with engagement (-1 to 1)"` + PostCount int `json:"postCount" jsonschema_description:"Number of posts containing this topic"` + SamplePosts []string `json:"samplePosts" jsonschema_description:"Sample posts (max 2)"` +} diff --git a/internal/server/webhook.go b/internal/server/webhook.go index 73b972a..75708c1 100644 --- a/internal/server/webhook.go +++ b/internal/server/webhook.go @@ -25,7 +25,7 @@ func newWebhookSender(cfg *config.Config, m *MetricsMiddleware) *WebhookSender { } return &WebhookSender{ - cfg: cfg, + cfg: cfg, metrics: m, httpCli: &http.Client{Timeout: 10 * time.Second}, server: cfg.MatrixServerID, @@ -38,7 +38,7 @@ func (wh *WebhookSender) Send(ctx context.Context, jsonBytes []byte) { return } - if jsonBytes == nil || len(jsonBytes) == 0 { + if len(jsonBytes) == 0 { log.ErrorWithContext(ctx, "webhook: empty payload ignored") return } diff --git a/internal/store/AnalyticsCache.sql.go b/internal/store/AnalyticsCache.sql.go new file mode 100644 index 0000000..8ee6cb0 --- /dev/null +++ b/internal/store/AnalyticsCache.sql.go @@ -0,0 +1,96 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: AnalyticsCache.sql + +package store + +import ( + "context" + "encoding/json" + "time" +) + +const deleteAnalyticsCache = `-- name: DeleteAnalyticsCache :exec +DELETE FROM analytics_cache +WHERE user_id = $1 AND platform = $2 AND tool_name = $3 AND time_range_hash = $4 +` + +type DeleteAnalyticsCacheParams struct { + UserID string + Platform string + ToolName string + TimeRangeHash string +} + +func (q *Queries) DeleteAnalyticsCache(ctx context.Context, arg DeleteAnalyticsCacheParams) error { + _, err := q.db.ExecContext(ctx, deleteAnalyticsCache, + arg.UserID, + arg.Platform, + arg.ToolName, + arg.TimeRangeHash, + ) + return err +} + +const getAnalyticsCache = `-- name: GetAnalyticsCache :one +SELECT result FROM analytics_cache +WHERE user_id = $1 AND platform = $2 AND tool_name = $3 AND time_range_hash = $4 + AND expires_at > now() +` + +type GetAnalyticsCacheParams struct { + UserID string + Platform string + ToolName string + TimeRangeHash string +} + +func (q *Queries) GetAnalyticsCache(ctx context.Context, arg GetAnalyticsCacheParams) (json.RawMessage, error) { + row := q.db.QueryRowContext(ctx, getAnalyticsCache, + arg.UserID, + arg.Platform, + arg.ToolName, + arg.TimeRangeHash, + ) + var result json.RawMessage + err := row.Scan(&result) + return result, err +} + +const pruneAnalyticsCache = `-- name: PruneAnalyticsCache :exec +DELETE FROM analytics_cache WHERE expires_at <= now() +` + +func (q *Queries) PruneAnalyticsCache(ctx context.Context) error { + _, err := q.db.ExecContext(ctx, pruneAnalyticsCache) + return err +} + +const upsertAnalyticsCache = `-- name: UpsertAnalyticsCache :exec +INSERT INTO analytics_cache (user_id, platform, tool_name, time_range_hash, result, expires_at) +VALUES ($1, $2, $3, $4, $5, $6) +ON CONFLICT (user_id, platform, tool_name, time_range_hash) +DO UPDATE SET result = EXCLUDED.result, expires_at = EXCLUDED.expires_at, computed_at = now() +` + +type UpsertAnalyticsCacheParams struct { + UserID string + Platform string + ToolName string + TimeRangeHash string + Result json.RawMessage + ExpiresAt time.Time +} + +func (q *Queries) UpsertAnalyticsCache(ctx context.Context, arg UpsertAnalyticsCacheParams) error { + _, err := q.db.ExecContext(ctx, upsertAnalyticsCache, + arg.UserID, + arg.Platform, + arg.ToolName, + arg.TimeRangeHash, + arg.Result, + arg.ExpiresAt, + ) + return err +} diff --git a/internal/store/AnalyticsCoverage.sql.go b/internal/store/AnalyticsCoverage.sql.go new file mode 100644 index 0000000..3f8e0d1 --- /dev/null +++ b/internal/store/AnalyticsCoverage.sql.go @@ -0,0 +1,59 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: AnalyticsCoverage.sql + +package store + +import ( + "context" + "time" +) + +const getAnalyticsCoverage = `-- name: GetAnalyticsCoverage :one +SELECT covered_start, covered_end FROM analytics_coverage +WHERE user_id = $1 AND platform = $2 +` + +type GetAnalyticsCoverageParams struct { + UserID string + Platform string +} + +type GetAnalyticsCoverageRow struct { + CoveredStart time.Time + CoveredEnd time.Time +} + +func (q *Queries) GetAnalyticsCoverage(ctx context.Context, arg GetAnalyticsCoverageParams) (GetAnalyticsCoverageRow, error) { + row := q.db.QueryRowContext(ctx, getAnalyticsCoverage, arg.UserID, arg.Platform) + var i GetAnalyticsCoverageRow + err := row.Scan(&i.CoveredStart, &i.CoveredEnd) + return i, err +} + +const upsertAnalyticsCoverage = `-- name: UpsertAnalyticsCoverage :exec +INSERT INTO analytics_coverage (user_id, platform, covered_start, covered_end, updated_at) +VALUES ($1, $2, $3, $4, now()) +ON CONFLICT (user_id, platform) +DO UPDATE SET covered_start = LEAST(analytics_coverage.covered_start, EXCLUDED.covered_start), + covered_end = GREATEST(analytics_coverage.covered_end, EXCLUDED.covered_end), + updated_at = now() +` + +type UpsertAnalyticsCoverageParams struct { + UserID string + Platform string + CoveredStart time.Time + CoveredEnd time.Time +} + +func (q *Queries) UpsertAnalyticsCoverage(ctx context.Context, arg UpsertAnalyticsCoverageParams) error { + _, err := q.db.ExecContext(ctx, upsertAnalyticsCoverage, + arg.UserID, + arg.Platform, + arg.CoveredStart, + arg.CoveredEnd, + ) + return err +} diff --git a/internal/store/AnalyticsPosts.sql.go b/internal/store/AnalyticsPosts.sql.go new file mode 100644 index 0000000..1cf59a1 --- /dev/null +++ b/internal/store/AnalyticsPosts.sql.go @@ -0,0 +1,113 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: AnalyticsPosts.sql + +package store + +import ( + "context" + "time" +) + +const getAnalyticsPostsInRange = `-- name: GetAnalyticsPostsInRange :many +SELECT post_id, post_text, post_timestamp, likes, reposts, replies, quotes +FROM analytics_posts +WHERE user_id = $1 AND platform = $2 AND post_timestamp >= $3 AND post_timestamp <= $4 +ORDER BY post_timestamp DESC +LIMIT $5 +` + +type GetAnalyticsPostsInRangeParams struct { + UserID string + Platform string + PostTimestamp time.Time + PostTimestamp_2 time.Time + Limit int32 +} + +type GetAnalyticsPostsInRangeRow struct { + PostID string + PostText string + PostTimestamp time.Time + Likes int64 + Reposts int64 + Replies int64 + Quotes int64 +} + +func (q *Queries) GetAnalyticsPostsInRange(ctx context.Context, arg GetAnalyticsPostsInRangeParams) ([]GetAnalyticsPostsInRangeRow, error) { + rows, err := q.db.QueryContext(ctx, getAnalyticsPostsInRange, + arg.UserID, + arg.Platform, + arg.PostTimestamp, + arg.PostTimestamp_2, + arg.Limit, + ) + if err != nil { + return nil, err + } + defer rows.Close() + var items []GetAnalyticsPostsInRangeRow + for rows.Next() { + var i GetAnalyticsPostsInRangeRow + if err := rows.Scan( + &i.PostID, + &i.PostText, + &i.PostTimestamp, + &i.Likes, + &i.Reposts, + &i.Replies, + &i.Quotes, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const upsertAnalyticsPosts = `-- name: UpsertAnalyticsPosts :exec +INSERT INTO analytics_posts (user_id, platform, post_id, post_text, post_timestamp, likes, reposts, replies, quotes) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) +ON CONFLICT (user_id, platform, post_id) +DO UPDATE SET post_text = EXCLUDED.post_text, + post_timestamp = EXCLUDED.post_timestamp, + likes = EXCLUDED.likes, + reposts = EXCLUDED.reposts, + replies = EXCLUDED.replies, + quotes = EXCLUDED.quotes +` + +type UpsertAnalyticsPostsParams struct { + UserID string + Platform string + PostID string + PostText string + PostTimestamp time.Time + Likes int64 + Reposts int64 + Replies int64 + Quotes int64 +} + +func (q *Queries) UpsertAnalyticsPosts(ctx context.Context, arg UpsertAnalyticsPostsParams) error { + _, err := q.db.ExecContext(ctx, upsertAnalyticsPosts, + arg.UserID, + arg.Platform, + arg.PostID, + arg.PostText, + arg.PostTimestamp, + arg.Likes, + arg.Reposts, + arg.Replies, + arg.Quotes, + ) + return err +} diff --git a/internal/store/DeleteStylePosts.sql.go b/internal/store/DeleteStylePosts.sql.go new file mode 100644 index 0000000..6a8ce7b --- /dev/null +++ b/internal/store/DeleteStylePosts.sql.go @@ -0,0 +1,25 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: DeleteStylePosts.sql + +package store + +import ( + "context" +) + +const deleteStylePosts = `-- name: DeleteStylePosts :exec +DELETE FROM user_style_posts +WHERE user_id = $1 AND platform = $2 AND platform != 'Twitter' +` + +type DeleteStylePostsParams struct { + UserID string + Platform string +} + +func (q *Queries) DeleteStylePosts(ctx context.Context, arg DeleteStylePostsParams) error { + _, err := q.db.ExecContext(ctx, deleteStylePosts, arg.UserID, arg.Platform) + return err +} diff --git a/internal/store/FindSimilarUsers.sql.go b/internal/store/FindSimilarUsers.sql.go index 638c23b..ba1678e 100644 --- a/internal/store/FindSimilarUsers.sql.go +++ b/internal/store/FindSimilarUsers.sql.go @@ -20,14 +20,14 @@ LIMIT $2 ` type FindSimilarUsersParams struct { - QueryEmbedding any + QueryEmbedding interface{} LimitVal int32 } type FindSimilarUsersRow struct { UserID string Platform string - Embedding any + Embedding interface{} TopKeywords []string StyleSummary string PostCount int32 diff --git a/internal/store/GetStylePostByID.sql.go b/internal/store/GetStylePostByID.sql.go new file mode 100644 index 0000000..2f6d77a --- /dev/null +++ b/internal/store/GetStylePostByID.sql.go @@ -0,0 +1,33 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: GetStylePostByID.sql + +package store + +import ( + "context" +) + +const getStylePostByID = `-- name: GetStylePostByID :one +SELECT post_id, post_text FROM user_style_posts +WHERE user_id = $1 AND platform = $2 AND post_id = $3 +` + +type GetStylePostByIDParams struct { + UserID string + Platform string + PostID string +} + +type GetStylePostByIDRow struct { + PostID string + PostText string +} + +func (q *Queries) GetStylePostByID(ctx context.Context, arg GetStylePostByIDParams) (GetStylePostByIDRow, error) { + row := q.db.QueryRowContext(ctx, getStylePostByID, arg.UserID, arg.Platform, arg.PostID) + var i GetStylePostByIDRow + err := row.Scan(&i.PostID, &i.PostText) + return i, err +} diff --git a/internal/store/GetStylePosts.sql.go b/internal/store/GetStylePosts.sql.go new file mode 100644 index 0000000..e8872d4 --- /dev/null +++ b/internal/store/GetStylePosts.sql.go @@ -0,0 +1,51 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: GetStylePosts.sql + +package store + +import ( + "context" +) + +const getStylePosts = `-- name: GetStylePosts :many +SELECT post_id, post_text FROM user_style_posts +WHERE user_id = $1 AND platform = $2 +ORDER BY created_at DESC +LIMIT $3 +` + +type GetStylePostsParams struct { + UserID string + Platform string + Limit int32 +} + +type GetStylePostsRow struct { + PostID string + PostText string +} + +func (q *Queries) GetStylePosts(ctx context.Context, arg GetStylePostsParams) ([]GetStylePostsRow, error) { + rows, err := q.db.QueryContext(ctx, getStylePosts, arg.UserID, arg.Platform, arg.Limit) + if err != nil { + return nil, err + } + defer rows.Close() + var items []GetStylePostsRow + for rows.Next() { + var i GetStylePostsRow + if err := rows.Scan(&i.PostID, &i.PostText); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} diff --git a/internal/store/GetStylePostsByIDs.sql.go b/internal/store/GetStylePostsByIDs.sql.go new file mode 100644 index 0000000..b1996ab --- /dev/null +++ b/internal/store/GetStylePostsByIDs.sql.go @@ -0,0 +1,51 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: GetStylePostsByIDs.sql + +package store + +import ( + "context" + + "github.com/lib/pq" +) + +const getStylePostsByIDs = `-- name: GetStylePostsByIDs :many +SELECT post_id, post_text FROM user_style_posts +WHERE user_id = $1 AND platform = $2 AND post_id = ANY($3::text[]) +` + +type GetStylePostsByIDsParams struct { + UserID string + Platform string + Column3 []string +} + +type GetStylePostsByIDsRow struct { + PostID string + PostText string +} + +func (q *Queries) GetStylePostsByIDs(ctx context.Context, arg GetStylePostsByIDsParams) ([]GetStylePostsByIDsRow, error) { + rows, err := q.db.QueryContext(ctx, getStylePostsByIDs, arg.UserID, arg.Platform, pq.Array(arg.Column3)) + if err != nil { + return nil, err + } + defer rows.Close() + var items []GetStylePostsByIDsRow + for rows.Next() { + var i GetStylePostsByIDsRow + if err := rows.Scan(&i.PostID, &i.PostText); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} diff --git a/internal/store/PruneStylePosts.sql.go b/internal/store/PruneStylePosts.sql.go new file mode 100644 index 0000000..52a36c8 --- /dev/null +++ b/internal/store/PruneStylePosts.sql.go @@ -0,0 +1,20 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: PruneStylePosts.sql + +package store + +import ( + "context" +) + +const pruneStylePosts = `-- name: PruneStylePosts :exec +DELETE FROM user_style_posts +WHERE created_at < now() - interval '7 days' AND platform != 'Twitter' +` + +func (q *Queries) PruneStylePosts(ctx context.Context) error { + _, err := q.db.ExecContext(ctx, pruneStylePosts) + return err +} diff --git a/internal/store/UpsertStylePosts.sql.go b/internal/store/UpsertStylePosts.sql.go new file mode 100644 index 0000000..8839ffe --- /dev/null +++ b/internal/store/UpsertStylePosts.sql.go @@ -0,0 +1,34 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: UpsertStylePosts.sql + +package store + +import ( + "context" +) + +const upsertStylePosts = `-- name: UpsertStylePosts :exec +INSERT INTO user_style_posts (user_id, platform, post_id, post_text) +VALUES ($1, $2, $3, $4) +ON CONFLICT (user_id, platform, post_id) +DO UPDATE SET post_text = EXCLUDED.post_text, created_at = now() +` + +type UpsertStylePostsParams struct { + UserID string + Platform string + PostID string + PostText string +} + +func (q *Queries) UpsertStylePosts(ctx context.Context, arg UpsertStylePostsParams) error { + _, err := q.db.ExecContext(ctx, upsertStylePosts, + arg.UserID, + arg.Platform, + arg.PostID, + arg.PostText, + ) + return err +} diff --git a/internal/store/UpsertStyleProfiles.sql.go b/internal/store/UpsertStyleProfiles.sql.go index 1a53dd3..7dadce2 100644 --- a/internal/store/UpsertStyleProfiles.sql.go +++ b/internal/store/UpsertStyleProfiles.sql.go @@ -26,7 +26,7 @@ DO UPDATE SET type UpsertStyleProfileParams struct { UserID string Platform string - Embedding any + Embedding interface{} TopKeywords []string StyleSummary string PostCount int32 diff --git a/internal/store/db.go b/internal/store/db.go index 6bf74de..fd993d9 100644 --- a/internal/store/db.go +++ b/internal/store/db.go @@ -10,10 +10,10 @@ import ( ) type DBTX interface { - ExecContext(context.Context, string, ...any) (sql.Result, error) + ExecContext(context.Context, string, ...interface{}) (sql.Result, error) PrepareContext(context.Context, string) (*sql.Stmt, error) - QueryContext(context.Context, string, ...any) (*sql.Rows, error) - QueryRowContext(context.Context, string, ...any) *sql.Row + QueryContext(context.Context, string, ...interface{}) (*sql.Rows, error) + QueryRowContext(context.Context, string, ...interface{}) *sql.Row } func New(db DBTX) *Queries { diff --git a/internal/store/integration_test.go b/internal/store/integration_test.go index 7f1cedf..820fafe 100644 --- a/internal/store/integration_test.go +++ b/internal/store/integration_test.go @@ -35,7 +35,7 @@ func TestStoreRoundTrip(t *testing.T) { q := New(conn) uid := fmt.Sprintf("test_%d", time.Now().UnixNano()) - emb := "[" + strings.TrimSuffix(strings.Repeat("0.0,", 768), ",") + "]" + emb := "[" + strings.TrimSuffix(strings.Repeat("0.0,", 1024), ",") + "]" if err := q.UpsertStyleProfile(ctx, UpsertStyleProfileParams{ UserID: uid, diff --git a/internal/store/models.go b/internal/store/models.go index c62215b..d3820d8 100644 --- a/internal/store/models.go +++ b/internal/store/models.go @@ -1,13 +1,48 @@ // Code generated by sqlc. DO NOT EDIT. // versions: -// sqlc v1.30.0 +// sqlc v1.31.1 package store import ( "database/sql" + "encoding/json" + "time" ) +type AnalyticsCache struct { + ID int64 + UserID string + Platform string + ToolName string + TimeRangeHash string + Result json.RawMessage + ComputedAt time.Time + ExpiresAt time.Time +} + +type AnalyticsCoverage struct { + UserID string + Platform string + CoveredStart time.Time + CoveredEnd time.Time + UpdatedAt time.Time +} + +type AnalyticsPost struct { + ID int64 + UserID string + Platform string + PostID string + PostText string + PostTimestamp time.Time + Likes int64 + Reposts int64 + Replies int64 + Quotes int64 + CreatedAt time.Time +} + type Transaction struct { ID int64 X402TransactionHash string @@ -19,11 +54,20 @@ type Transaction struct { CreatedAt sql.NullTime } +type UserStylePost struct { + ID int64 + UserID string + Platform string + PostID string + PostText string + CreatedAt time.Time +} + type UserStyleProfile struct { ID int64 UserID string Platform string - Embedding any + Embedding interface{} TopKeywords []string StyleSummary string PostCount int32 diff --git a/internal/utils/styleProfiles.go b/internal/utils/styleProfiles.go index 27c372d..7a6a587 100644 --- a/internal/utils/styleProfiles.go +++ b/internal/utils/styleProfiles.go @@ -38,11 +38,9 @@ func ParseEmbedding(value any) ([]float64, error) { return vec, nil case string: - // pgvector returns a string like "[0.12, 0.53, ...]" sometimes return parseVectorString(v) case []byte: - // lib/pq returns unknown-type columns (e.g. vector) as []byte return parseVectorString(string(v)) default: @@ -76,7 +74,7 @@ func AverageVector(vs [][]float64) []float64 { func TopWords(posts []string, stopWords map[string]struct{}, limit int) []string { counts := map[string]int{} for _, p := range posts { - for _, w := range strings.Fields(strings.ToLower(p)) { + for w := range strings.FieldsSeq(strings.ToLower(p)) { if len(w) < 4 { continue } diff --git a/scripts/fund-wallet.sh b/scripts/fund-wallet.sh new file mode 100755 index 0000000..7ad808d --- /dev/null +++ b/scripts/fund-wallet.sh @@ -0,0 +1,62 @@ +#!/bin/bash +set -euo pipefail + +echo "=== Funding Wallet with Base Sepolia USDC ===" + +# Required environment variables +: "${PRIVATE_KEY:?PRIVATE_KEY must be set}" +: "${FUNDING_FAUCET_URL:?FUNDING_FAUCET_URL must be set}" + +WALLET_ADDRESS=$(cast wallet address --private-key "$PRIVATE_KEY") +echo "Wallet address: $WALLET_ADDRESS" + +# Check current USDC balance +USDC_CONTRACT="0x036CbD53842c5426634e7929541eC2318f3dCF7e" +RPC_URL="https://sepolia.base.org" + +BALANCE=$(cast call "$USDC_CONTRACT" "balanceOf(address)(uint256)" "$WALLET_ADDRESS" --rpc-url "$RPC_URL" 2>/dev/null || echo "0") +BALANCE_DECIMAL=$(cast to-dec "$BALANCE" 2>/dev/null || echo "0") +BALANCE_USDC=$(echo "scale=6; $BALANCE_DECIMAL / 1000000" | bc 2>/dev/null || echo "0") + +echo "Current USDC balance: $BALANCE_USDC USDC" + +# If balance is less than 10 USDC, request from faucet +if (( $(echo "$BALANCE_USDC < 10" | bc -l) )); then + echo "Balance below 10 USDC, requesting from faucet..." + + # Request from faucet (Coinbase CDP faucet API format) + curl -X POST "$FUNDING_FAUCET_URL" \ + -H "Content-Type: application/json" \ + -d "{\"address\": \"$WALLET_ADDRESS\", \"token\": \"USDC\"}" \ + || echo "Faucet request failed, may need manual funding" + + # Wait for funding + echo "Waiting for funding..." + sleep 30 + + # Check balance again + BALANCE=$(cast call "$USDC_CONTRACT" "balanceOf(address)(uint256)" "$WALLET_ADDRESS" --rpc-url "$RPC_URL" 2>/dev/null || echo "0") + BALANCE_DECIMAL=$(cast to-dec "$BALANCE" 2>/dev/null || echo "0") + BALANCE_USDC=$(echo "scale=6; $BALANCE_DECIMAL / 1000000" | bc 2>/dev/null || echo "0") + echo "New USDC balance: $BALANCE_USDC USDC" +else + echo "Sufficient USDC balance" +fi + +# Also fund with ETH for gas +ETH_BALANCE=$(cast balance "$WALLET_ADDRESS" --rpc-url "$RPC_URL" 2>/dev/null || echo "0") +ETH_BALANCE_DECIMAL=$(cast to-dec "$ETH_BALANCE" 2>/dev/null || echo "0") +ETH_BALANCE_ETH=$(echo "scale=6; $ETH_BALANCE_DECIMAL / 1000000000000000000" | bc 2>/dev/null || echo "0") + +echo "Current ETH balance: $ETH_BALANCE_ETH ETH" + +if (( $(echo "$ETH_BALANCE_ETH < 0.01" | bc -l) )); then + echo "ETH balance low, requesting from faucet..." + curl -X POST "$FUNDING_FAUCET_URL" \ + -H "Content-Type: application/json" \ + -d "{\"address\": \"$WALLET_ADDRESS\", \"token\": \"ETH\"}" \ + || echo "ETH faucet request failed" + sleep 10 +fi + +echo "=== Wallet Funding Complete ===" \ No newline at end of file diff --git a/scripts/start-test-server.sh b/scripts/start-test-server.sh new file mode 100755 index 0000000..cdee795 --- /dev/null +++ b/scripts/start-test-server.sh @@ -0,0 +1,83 @@ +#!/bin/bash +set -euo pipefail + +echo "=== Starting Test Server ===" + +# Required environment variables +: "${PRIVATE_KEY:?PRIVATE_KEY must be set}" +: "${X402_WALLET_ADDRESS:?X402_WALLET_ADDRESS must be set}" + +# Set up environment for local testing +export SERVER_ADDRESS="127.0.0.1" +export SERVER_PORT="8080" +export GOOSE_DBSTRING="postgres://postgres:postgres@localhost:5432/mcpserver?sslmode=disable" +export ACTIVE_NETWORKS="sepolia" +export X402_FACILITATOR_URL="https://api.cdp.coinbase.com/platform/v2/x402" +export X402_WALLET_ADDRESS="$X402_WALLET_ADDRESS" +export OLLAMA_HOST="http://localhost:11434" +export MIGRATE_ON_START="true" +export MIGRATIONS_DIR="sql/schema" +export READYZ_PORT="9090" +export METRICS_PORT="9091" +export LOG_FILE="/tmp/mcpserver-test.log" +export LOG_LEVEL="debug" + +# Use test credentials or empty for local testing +export BLUESKY_DID="${BLUESKY_DID:-}" +export BLUESKY_APP_PASSWORD="${BLUESKY_APP_PASSWORD:-}" +export BLUESKY_HOST="${BLUESKY_HOST:-https://bsky.social}" +export AP_HOST="${AP_HOST:-https://mastodon.social}" +export AP_ACCESS_TOKEN="${AP_ACCESS_TOKEN:-}" +export NOSTR_RELAYS="${NOSTR_RELAYS:-wss://relay.damus.io,wss://nos.lol}" +export TWITTER_BEARER_TOKEN="${TWITTER_BEARER_TOKEN:-}" +export CDP_API_KEY_ID="${CDP_API_KEY_ID:-}" +export CDP_API_SECRET="${CDP_API_SECRET:-}" +export ENRICH_CROSS_PLATFORM="false" +export PREWARM_AUTHORS="" +export PREWARM_FEED_URI="" + +# Start the server in background +echo "Starting server on port 8080..." +go run ./cmd/server/main.go & +SERVER_PID=$! + +# Save PID for cleanup +echo $SERVER_PID > /tmp/test-server.pid + +# Wait for server to be ready +echo "Waiting for server to be ready..." +for i in {1..30}; do + if curl -sf http://localhost:9090/readyz >/dev/null 2>&1; then + echo "Server is ready" + break + fi + sleep 1 +done + +# Verify server is responding +curl -sf http://localhost:8080/healthz >/dev/null 2>&1 || { + echo "Server health check failed" + kill $SERVER_PID 2>/dev/null || true + exit 1 +} + +# Verify payment endpoints are working (should return 402 without payment) +echo "Verifying payment endpoints..." +if curl -sf -X POST http://localhost:8080/v1/getActor \ + -H "Content-Type: application/json" \ + -d '{"author": "did:plc:z72i7hdynmk6r22z27h6tvur"}' >/dev/null 2>&1; then + echo "WARNING: Payable endpoint getActor returned 200 without payment!" +else + echo "Payment endpoint getActor correctly requires payment (402)" +fi + +# Verify free endpoint works +if curl -sf -X POST http://localhost:8080/v1/getFeed \ + -H "Content-Type: application/json" \ + -d '{"uri": "https://bsky.app/profile/did:plc:z72i7hdynmk6r22z27h6tvur/feed/whats-hot"}' >/dev/null 2>&1; then + echo "Free endpoint getFeed works correctly" +else + echo "WARNING: Free endpoint getFeed failed!" +fi + +echo "=== Test Server Started (PID: $SERVER_PID) ===" \ No newline at end of file diff --git a/sql/queries/AnalyticsCache.sql b/sql/queries/AnalyticsCache.sql new file mode 100644 index 0000000..670da7e --- /dev/null +++ b/sql/queries/AnalyticsCache.sql @@ -0,0 +1,17 @@ +-- name: GetAnalyticsCache :one +SELECT result FROM analytics_cache +WHERE user_id = $1 AND platform = $2 AND tool_name = $3 AND time_range_hash = $4 + AND expires_at > now(); + +-- name: UpsertAnalyticsCache :exec +INSERT INTO analytics_cache (user_id, platform, tool_name, time_range_hash, result, expires_at) +VALUES ($1, $2, $3, $4, $5, $6) +ON CONFLICT (user_id, platform, tool_name, time_range_hash) +DO UPDATE SET result = EXCLUDED.result, expires_at = EXCLUDED.expires_at, computed_at = now(); + +-- name: DeleteAnalyticsCache :exec +DELETE FROM analytics_cache +WHERE user_id = $1 AND platform = $2 AND tool_name = $3 AND time_range_hash = $4; + +-- name: PruneAnalyticsCache :exec +DELETE FROM analytics_cache WHERE expires_at <= now(); \ No newline at end of file diff --git a/sql/queries/AnalyticsCoverage.sql b/sql/queries/AnalyticsCoverage.sql new file mode 100644 index 0000000..58ad395 --- /dev/null +++ b/sql/queries/AnalyticsCoverage.sql @@ -0,0 +1,11 @@ +-- name: GetAnalyticsCoverage :one +SELECT covered_start, covered_end FROM analytics_coverage +WHERE user_id = $1 AND platform = $2; + +-- name: UpsertAnalyticsCoverage :exec +INSERT INTO analytics_coverage (user_id, platform, covered_start, covered_end, updated_at) +VALUES ($1, $2, $3, $4, now()) +ON CONFLICT (user_id, platform) +DO UPDATE SET covered_start = LEAST(analytics_coverage.covered_start, EXCLUDED.covered_start), + covered_end = GREATEST(analytics_coverage.covered_end, EXCLUDED.covered_end), + updated_at = now(); diff --git a/sql/queries/AnalyticsPosts.sql b/sql/queries/AnalyticsPosts.sql new file mode 100644 index 0000000..6f7c31e --- /dev/null +++ b/sql/queries/AnalyticsPosts.sql @@ -0,0 +1,17 @@ +-- name: GetAnalyticsPostsInRange :many +SELECT post_id, post_text, post_timestamp, likes, reposts, replies, quotes +FROM analytics_posts +WHERE user_id = $1 AND platform = $2 AND post_timestamp >= $3 AND post_timestamp <= $4 +ORDER BY post_timestamp DESC +LIMIT $5; + +-- name: UpsertAnalyticsPosts :exec +INSERT INTO analytics_posts (user_id, platform, post_id, post_text, post_timestamp, likes, reposts, replies, quotes) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) +ON CONFLICT (user_id, platform, post_id) +DO UPDATE SET post_text = EXCLUDED.post_text, + post_timestamp = EXCLUDED.post_timestamp, + likes = EXCLUDED.likes, + reposts = EXCLUDED.reposts, + replies = EXCLUDED.replies, + quotes = EXCLUDED.quotes; diff --git a/sql/queries/DeleteStylePosts.sql b/sql/queries/DeleteStylePosts.sql new file mode 100644 index 0000000..163eaae --- /dev/null +++ b/sql/queries/DeleteStylePosts.sql @@ -0,0 +1,3 @@ +-- name: DeleteStylePosts :exec +DELETE FROM user_style_posts +WHERE user_id = $1 AND platform = $2 AND platform != 'Twitter'; \ No newline at end of file diff --git a/sql/queries/GetStylePostByID.sql b/sql/queries/GetStylePostByID.sql new file mode 100644 index 0000000..255531c --- /dev/null +++ b/sql/queries/GetStylePostByID.sql @@ -0,0 +1,3 @@ +-- name: GetStylePostByID :one +SELECT post_id, post_text FROM user_style_posts +WHERE user_id = $1 AND platform = $2 AND post_id = $3; \ No newline at end of file diff --git a/sql/queries/GetStylePosts.sql b/sql/queries/GetStylePosts.sql new file mode 100644 index 0000000..3c1ca49 --- /dev/null +++ b/sql/queries/GetStylePosts.sql @@ -0,0 +1,5 @@ +-- name: GetStylePosts :many +SELECT post_id, post_text FROM user_style_posts +WHERE user_id = $1 AND platform = $2 +ORDER BY created_at DESC +LIMIT $3; \ No newline at end of file diff --git a/sql/queries/GetStylePostsByIDs.sql b/sql/queries/GetStylePostsByIDs.sql new file mode 100644 index 0000000..932b794 --- /dev/null +++ b/sql/queries/GetStylePostsByIDs.sql @@ -0,0 +1,3 @@ +-- name: GetStylePostsByIDs :many +SELECT post_id, post_text FROM user_style_posts +WHERE user_id = $1 AND platform = $2 AND post_id = ANY($3::text[]); \ No newline at end of file diff --git a/sql/queries/PruneStylePosts.sql b/sql/queries/PruneStylePosts.sql new file mode 100644 index 0000000..064140d --- /dev/null +++ b/sql/queries/PruneStylePosts.sql @@ -0,0 +1,3 @@ +-- name: PruneStylePosts :exec +DELETE FROM user_style_posts +WHERE created_at < now() - interval '7 days' AND platform != 'Twitter'; \ No newline at end of file diff --git a/sql/queries/UpsertStylePosts.sql b/sql/queries/UpsertStylePosts.sql new file mode 100644 index 0000000..2c669bf --- /dev/null +++ b/sql/queries/UpsertStylePosts.sql @@ -0,0 +1,5 @@ +-- name: UpsertStylePosts :exec +INSERT INTO user_style_posts (user_id, platform, post_id, post_text) +VALUES ($1, $2, $3, $4) +ON CONFLICT (user_id, platform, post_id) +DO UPDATE SET post_text = EXCLUDED.post_text, created_at = now(); \ No newline at end of file diff --git a/sql/schema/001_transactions.sql b/sql/schema/001_transactions.sql index 700b6b8..612215f 100644 --- a/sql/schema/001_transactions.sql +++ b/sql/schema/001_transactions.sql @@ -1,5 +1,5 @@ -- +goose up -CREATE TABLE transactions ( +CREATE TABLE IF NOT EXISTS transactions ( id BIGSERIAL PRIMARY KEY, x402_transaction_hash VARCHAR(255) NOT NULL UNIQUE, amount NUMERIC NOT NULL, diff --git a/sql/schema/002_user_style_profiles.sql b/sql/schema/002_user_style_profiles.sql index 14d6307..6241a7e 100644 --- a/sql/schema/002_user_style_profiles.sql +++ b/sql/schema/002_user_style_profiles.sql @@ -1,11 +1,11 @@ -- +goose up CREATE EXTENSION IF NOT EXISTS vector; -CREATE TABLE user_style_profiles ( +CREATE TABLE IF NOT EXISTS user_style_profiles ( id BIGSERIAL PRIMARY KEY, user_id TEXT NOT NULL, platform TEXT NOT NULL, - embedding VECTOR(768) NOT NULL, + embedding VECTOR(1024) NOT NULL, top_keywords TEXT[] NOT NULL, style_summary TEXT NOT NULL, post_count INTEGER NOT NULL, diff --git a/sql/schema/003_add_cache_version.sql b/sql/schema/003_add_cache_version.sql index 42172d6..4c86c15 100644 --- a/sql/schema/003_add_cache_version.sql +++ b/sql/schema/003_add_cache_version.sql @@ -4,7 +4,7 @@ ALTER TABLE user_style_profiles ADD COLUMN schema_version INTEGER NOT NULL DEFAULT 1, ADD COLUMN algorithm_version INTEGER NOT NULL DEFAULT 1; -CREATE INDEX idx_user_style_profiles_versions +CREATE INDEX IF NOT EXISTS idx_user_style_profiles_versions ON user_style_profiles (model_version, schema_version, algorithm_version); -- +goose Down diff --git a/sql/schema/004_add_style_posts.sql b/sql/schema/004_add_style_posts.sql new file mode 100644 index 0000000..cc020c1 --- /dev/null +++ b/sql/schema/004_add_style_posts.sql @@ -0,0 +1,17 @@ +-- +goose up +CREATE TABLE IF NOT EXISTS user_style_posts ( + id BIGSERIAL PRIMARY KEY, + user_id TEXT NOT NULL, + platform TEXT NOT NULL, + post_id TEXT NOT NULL, + post_text TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE (user_id, platform, post_id) +); + +CREATE INDEX idx_user_style_posts_user_platform + ON user_style_posts (user_id, platform); + +-- +goose down +DROP INDEX IF EXISTS idx_user_style_posts_user_platform; +DROP TABLE IF EXISTS user_style_posts; \ No newline at end of file diff --git a/sql/schema/005_add_analytics_cache.sql b/sql/schema/005_add_analytics_cache.sql new file mode 100644 index 0000000..3dd106c --- /dev/null +++ b/sql/schema/005_add_analytics_cache.sql @@ -0,0 +1,23 @@ +-- +goose up +CREATE TABLE IF NOT EXISTS analytics_cache ( + id BIGSERIAL PRIMARY KEY, + user_id TEXT NOT NULL, + platform TEXT NOT NULL, + tool_name TEXT NOT NULL, + time_range_hash TEXT NOT NULL, + result JSONB NOT NULL, + computed_at TIMESTAMPTZ NOT NULL DEFAULT now(), + expires_at TIMESTAMPTZ NOT NULL, + UNIQUE (user_id, platform, tool_name, time_range_hash) +); + +CREATE INDEX IF NOT EXISTS idx_analytics_cache_lookup + ON analytics_cache (user_id, platform, tool_name, time_range_hash); + +CREATE INDEX IF NOT EXISTS idx_analytics_cache_expires + ON analytics_cache (expires_at); + +-- +goose down +DROP INDEX IF EXISTS idx_analytics_cache_expires; +DROP INDEX IF EXISTS idx_analytics_cache_lookup; +DROP TABLE IF EXISTS analytics_cache; \ No newline at end of file diff --git a/sql/schema/006_add_analytics_posts.sql b/sql/schema/006_add_analytics_posts.sql new file mode 100644 index 0000000..36fc965 --- /dev/null +++ b/sql/schema/006_add_analytics_posts.sql @@ -0,0 +1,22 @@ +-- +goose up +CREATE TABLE IF NOT EXISTS analytics_posts ( + id BIGSERIAL PRIMARY KEY, + user_id TEXT NOT NULL, + platform TEXT NOT NULL, + post_id TEXT NOT NULL, + post_text TEXT NOT NULL, + post_timestamp TIMESTAMPTZ NOT NULL, + likes BIGINT NOT NULL DEFAULT 0, + reposts BIGINT NOT NULL DEFAULT 0, + replies BIGINT NOT NULL DEFAULT 0, + quotes BIGINT NOT NULL DEFAULT 0, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE (user_id, platform, post_id) +); + +CREATE INDEX IF NOT EXISTS idx_analytics_posts_user_platform_time + ON analytics_posts (user_id, platform, post_timestamp); + +-- +goose down +DROP INDEX IF EXISTS idx_analytics_posts_user_platform_time; +DROP TABLE IF EXISTS analytics_posts; diff --git a/sql/schema/007_add_analytics_coverage.sql b/sql/schema/007_add_analytics_coverage.sql new file mode 100644 index 0000000..b22ac24 --- /dev/null +++ b/sql/schema/007_add_analytics_coverage.sql @@ -0,0 +1,12 @@ +-- +goose up +CREATE TABLE IF NOT EXISTS analytics_coverage ( + user_id TEXT NOT NULL, + platform TEXT NOT NULL, + covered_start TIMESTAMPTZ NOT NULL, + covered_end TIMESTAMPTZ NOT NULL, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (user_id, platform) +); + +-- +goose down +DROP TABLE IF EXISTS analytics_coverage; diff --git a/testClient/main.go b/testClient/main.go index 394d2c1..2e77c1f 100644 --- a/testClient/main.go +++ b/testClient/main.go @@ -1,16 +1,22 @@ package main import ( - "bytes" "context" + "encoding/json" "fmt" "io" "log" - "os" - + "math/big" "net/http" "net/http/httputil" + "os" + "strings" + "time" + "github.com/ethereum/go-ethereum" + "github.com/ethereum/go-ethereum/accounts/abi" + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/ethclient" "github.com/joho/godotenv" x402 "github.com/x402-foundation/x402/go/v2" @@ -19,6 +25,15 @@ import ( evmsigners "github.com/x402-foundation/x402/go/v2/signers/evm" ) +const ( + USDCContractAddress = "0x036CbD53842c5426634e7929541eC2318f3dCF7e" + BaseSepoliaRPC = "https://sepolia.base.org" + MinUSDCRequired = 10.0 + MinETHRequired = 0.01 +) + +var usdcABI = `[{"constant":true,"inputs":[{"name":"_owner","type":"address"}],"name":"balanceOf","outputs":[{"name":"balance","type":"uint256"}],"type":"function"}]` + type LoggingRoundTripper struct { Proxied http.RoundTripper } @@ -31,7 +46,6 @@ func (lrt *LoggingRoundTripper) RoundTrip(req *http.Request) (*http.Response, er req.Body = rc } - // Dump request reqDump, err := httputil.DumpRequestOut(req, true) if err != nil { log.Printf("Error dumping request: %v", err) @@ -39,13 +53,11 @@ func (lrt *LoggingRoundTripper) RoundTrip(req *http.Request) (*http.Response, er fmt.Printf("\n--- REQUEST ---\n%s\n----------------\n", reqDump) } - // Perform request resp, err := lrt.Proxied.RoundTrip(req) if err != nil { return nil, err } - // Dump response respDump, err := httputil.DumpResponse(resp, true) if err != nil { log.Printf("Error dumping response: %v", err) @@ -56,13 +68,206 @@ func (lrt *LoggingRoundTripper) RoundTrip(req *http.Request) (*http.Response, er return resp, nil } +func callEndpoint(client *http.Client, method, endpoint, body string) error { + baseURL := getEnv("TEST_SERVER_URL", "http://localhost:8080") + url := fmt.Sprintf("%s/v1/%s", baseURL, endpoint) + req, err := http.NewRequest(method, url, strings.NewReader(body)) + if err != nil { + return fmt.Errorf("failed to create %s request: %w", endpoint, err) + } + req.Header.Set("Content-Type", "application/json") + + resp, err := client.Do(req) + if err != nil { + return fmt.Errorf("%s HTTP request failed: %w", endpoint, err) + } + defer resp.Body.Close() + + respBody, _ := io.ReadAll(resp.Body) + fmt.Printf("%s Response Status: %s\n", endpoint, resp.Status) + if len(respBody) > 0 { + fmt.Printf("%s Response Body: %s\n", endpoint, string(respBody)) + } + + if resp.StatusCode == 402 { + return fmt.Errorf("%s: payment required (402)", endpoint) + } + + return nil +} + +func getEnv(key, fallback string) string { + if value, ok := os.LookupEnv(key); ok { + return value + } + return fallback +} + +func getUSDCBalance(client *ethclient.Client, address common.Address) (*big.Int, error) { + contractAddr := common.HexToAddress(USDCContractAddress) + parsedABI, err := abi.JSON(strings.NewReader(usdcABI)) + if err != nil { + return nil, fmt.Errorf("failed to parse USDC ABI: %w", err) + } + data, err := parsedABI.Pack("balanceOf", address) + if err != nil { + return nil, fmt.Errorf("failed to pack balanceOf call: %w", err) + } + msg := ethereum.CallMsg{ + To: &contractAddr, + Data: data, + } + result, err := client.CallContract(context.Background(), msg, nil) + if err != nil { + return nil, fmt.Errorf("failed to call balanceOf: %w", err) + } + unpacked, err := parsedABI.Unpack("balanceOf", result) + if err != nil { + return nil, fmt.Errorf("failed to unpack balance: %w", err) + } + if len(unpacked) == 0 { + return nil, fmt.Errorf("no balance returned") + } + balance, ok := unpacked[0].(*big.Int) + if !ok { + return nil, fmt.Errorf("unexpected balance type: %T", unpacked[0]) + } + return balance, nil +} + +func getETHBalance(client *ethclient.Client, address common.Address) (*big.Int, error) { + balance, err := client.BalanceAt(context.Background(), address, nil) + if err != nil { + return nil, fmt.Errorf("failed to get ETH balance: %w", err) + } + return balance, nil +} + +func requestFaucetFunds(faucetURL, address, token string) error { + payload := map[string]string{ + "address": address, + "token": token, + } + jsonData, err := json.Marshal(payload) + if err != nil { + return fmt.Errorf("failed to marshal faucet request: %w", err) + } + resp, err := http.Post(faucetURL, "application/json", strings.NewReader(string(jsonData))) + if err != nil { + return fmt.Errorf("faucet request failed: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode >= 400 { + body, _ := io.ReadAll(resp.Body) + return fmt.Errorf("faucet returned error %d: %s", resp.StatusCode, string(body)) + } + return nil +} + +func ensureFundedWallet(privateKey, faucetURL, rpcURL string) error { + client, err := ethclient.Dial(rpcURL) + if err != nil { + return fmt.Errorf("failed to connect to RPC: %w", err) + } + defer client.Close() + + signer, err := evmsigners.NewClientSignerFromPrivateKey(privateKey) + if err != nil { + return fmt.Errorf("failed to create signer: %w", err) + } + addrStr := signer.Address() + address := common.HexToAddress(addrStr) + + log.Printf("Checking wallet balance for %s...", addrStr) + + usdcBalance, err := getUSDCBalance(client, address) + if err != nil { + return fmt.Errorf("failed to get USDC balance: %w", err) + } + usdcBalanceFloat := new(big.Float).Quo(new(big.Float).SetInt(usdcBalance), big.NewFloat(1e6)) + usdcFloat, _ := usdcBalanceFloat.Float64() + log.Printf("Current USDC balance: %.6f USDC", usdcFloat) + + ethBalance, err := getETHBalance(client, address) + if err != nil { + return fmt.Errorf("failed to get ETH balance: %w", err) + } + ethBalanceFloat := new(big.Float).Quo(new(big.Float).SetInt(ethBalance), big.NewFloat(1e18)) + ethFloat, _ := ethBalanceFloat.Float64() + log.Printf("Current ETH balance: %.6f ETH", ethFloat) + + needsFunding := false + if usdcFloat < MinUSDCRequired { + log.Printf("USDC balance %.6f below minimum %.2f, requesting funding...", usdcFloat, MinUSDCRequired) + needsFunding = true + } + if ethFloat < MinETHRequired { + log.Printf("ETH balance %.6f below minimum %.4f, requesting funding...", ethFloat, MinETHRequired) + needsFunding = true + } + + if needsFunding { + if faucetURL == "" { + return fmt.Errorf("FUNDING_FAUCET_URL not set but wallet needs funding") + } + if usdcFloat < MinUSDCRequired { + log.Printf("Requesting USDC from faucet...") + if err := requestFaucetFunds(faucetURL, addrStr, "USDC"); err != nil { + return fmt.Errorf("failed to request USDC from faucet: %w", err) + } + } + if ethFloat < MinETHRequired { + log.Printf("Requesting ETH from faucet...") + if err := requestFaucetFunds(faucetURL, addrStr, "ETH"); err != nil { + return fmt.Errorf("failed to request ETH from faucet: %w", err) + } + } + log.Printf("Waiting for funding to arrive...") + time.Sleep(30 * time.Second) + + usdcBalance, err = getUSDCBalance(client, address) + if err != nil { + return fmt.Errorf("failed to get USDC balance after funding: %w", err) + } + usdcBalanceFloat = new(big.Float).Quo(new(big.Float).SetInt(usdcBalance), big.NewFloat(1e6)) + usdcFloat, _ = usdcBalanceFloat.Float64() + log.Printf("New USDC balance: %.6f USDC", usdcFloat) + + ethBalance, err = getETHBalance(client, address) + if err != nil { + return fmt.Errorf("failed to get ETH balance after funding: %w", err) + } + ethBalanceFloat = new(big.Float).Quo(new(big.Float).SetInt(ethBalance), big.NewFloat(1e18)) + ethFloat, _ = ethBalanceFloat.Float64() + log.Printf("New ETH balance: %.6f ETH", ethFloat) + + if usdcFloat < MinUSDCRequired { + return fmt.Errorf("USDC balance still insufficient after funding: %.6f < %.2f", usdcFloat, MinUSDCRequired) + } + if ethFloat < MinETHRequired { + return fmt.Errorf("ETH balance still insufficient after funding: %.6f < %.4f", ethFloat, MinETHRequired) + } + } else { + log.Printf("Wallet has sufficient funds") + } + return nil +} + func main() { godotenv.Load() - var ( - privateKey = os.Getenv("PRIVATE_KEY") - serverURL = "https://api.naco.li/v1" - ctx = context.Background() - ) + privateKey := os.Getenv("PRIVATE_KEY") + serverURL := getEnv("TEST_SERVER_URL", "http://localhost:8080") + faucetURL := getEnv("FUNDING_FAUCET_URL", "") + rpcURL := getEnv("RPC_URL", BaseSepoliaRPC) + + if privateKey == "" { + log.Fatal("PRIVATE_KEY environment variable is required") + } + + log.Printf("Ensuring wallet is funded...") + if err := ensureFundedWallet(privateKey, faucetURL, rpcURL); err != nil { + log.Fatal("Failed to ensure wallet is funded: ", err) + } signer, err := evmsigners.NewClientSignerFromPrivateKey(privateKey) if err != nil { @@ -70,12 +275,13 @@ func main() { } x402Client := x402.Newx402Client(). - Register("eip155:*", evm.NewExactEvmScheme(signer, nil)) + Register("eip155:84532", evm.NewExactEvmScheme(signer, nil)) debugClient := &http.Client{ Transport: &LoggingRoundTripper{ Proxied: http.DefaultTransport, }, + Timeout: 60 * time.Second, } httpClient := x402http.WrapHTTPClientWithPayment( @@ -83,39 +289,44 @@ func main() { x402http.Newx402HTTPClient(x402Client), ) - log.Println("Starting REST API tests...") + log.Printf("Starting REST API endpoint tests against %s...", serverURL) - log.Println("\nCalling getFeed (free tool) via HTTP client...") - bodyGetFeed := []byte(`{"uri": "https://bsky.app/profile/did:plc:z72i7hdynmk6r22z27h6tvur/feed/whats-hot"}`) - resp, err := httpClient.Post(fmt.Sprintf("%s/getFeed", serverURL), "application/json", bytes.NewReader(bodyGetFeed)) - if err != nil { - log.Println("getFeed HTTP request failed:", err) - } else { - defer resp.Body.Close() - body, _ := io.ReadAll(resp.Body) - log.Printf("getFeed Response Status: %s\n", resp.Status) - log.Printf("getFeed Response Body: %s\n", string(body)) + testCases := []struct { + name string + method string + endpoint string + body string + }{ + {"getFeed", "POST", "getFeed", `{"uri": "https://bsky.app/profile/did:plc:z72i7hdynmk6r22z27h6tvur/feed/whats-hot"}`}, + {"getActor", "POST", "getActor", `{"author": "did:plc:z72i7hdynmk6r22z27h6tvur"}`}, + {"searchActors", "POST", "searchActors", `{"query": "test", "limit": 5}`}, + {"getPost", "POST", "getPost", `{"actor":"sharkgamiart.bsky.social","rkey":"3m47y27c67c2b"}`}, + {"getLatestPosts", "POST", "getLatestPosts", `{"author":"sharkgamiart.bsky.social","limit": 5}`}, + {"getPostsInTime", "POST", "getPostsInTime", `{"author":"sharkgamiart.bsky.social","startTime": "2024-01-01T00:00:00Z","endTime": "2024-12-31T23:59:59Z","limit": 5}`}, + {"getStyleProfile", "POST", "getStyleProfile", `{"author":"did:plc:z72i7hdynmk6r22z27h6tvur"}`}, + {"findSimilarUsers", "POST", "findSimilarUsers", `{"author":"did:plc:z72i7hdynmk6r22z27h6tvur","limit": 5}`}, } - log.Println("\nCalling getPost (gated paid tool) via HTTP client...") - bodyGetPost := []byte(`{"actor":"sharkgamiart.bsky.social","rkey":"3m47y27c67c2b"}`) + passed := 0 + failed := 0 - req, err := http.NewRequestWithContext(ctx, "POST", fmt.Sprintf("%s/getPost", serverURL), bytes.NewReader(bodyGetPost)) - if err != nil { - log.Fatal("Failed to create getPost request:", err) + for _, tc := range testCases { + fmt.Printf("\n=== Testing %s ===\n", tc.name) + err := callEndpoint(httpClient, tc.method, tc.endpoint, tc.body) + if err != nil { + log.Printf("FAIL %s: %v\n", tc.name, err) + failed++ + } else { + fmt.Printf("PASS %s\n", tc.name) + passed++ + } } - req.GetBody = func() (io.ReadCloser, error) { - return io.NopCloser(bytes.NewReader(bodyGetPost)), nil - } - req.Header.Set("Content-Type", "application/json") - respPost, err := httpClient.Do(req) - if err != nil { - log.Println("getPost HTTP request failed:", err) - } else { - defer respPost.Body.Close() - body, _ := io.ReadAll(respPost.Body) - log.Printf("getPost Response Status: %s\n", respPost.Status) - log.Printf("getPost Response Body: %s\n", string(body)) + log.Printf("\n=== Test Summary ===") + log.Printf("Passed: %d", passed) + log.Printf("Failed: %d", failed) + + if failed > 0 { + os.Exit(1) } }