diff --git a/.forgejo/workflows/test-endpoints.yaml b/.forgejo/workflows/test-endpoints.yaml deleted file mode 100644 index f47b513..0000000 --- a/.forgejo/workflows/test-endpoints.yaml +++ /dev/null @@ -1,142 +0,0 @@ -name: Test Every Endpoint on Base Sepolia - -on: - push: - branches: [ main, dev ] - pull_request: - branches: [ main ] - workflow_dispatch: - -env: - GO_VERSION: '1.25' - POSTGRES_DB: mcpserver - POSTGRES_USER: postgres - POSTGRES_PASSWORD: postgres - TEST_SERVER_URL: http://localhost:8080 - RPC_URL: https://sepolia.base.org - X402_FACILITATOR_URL: https://api.cdp.coinbase.com/platform/v2/x402 - -jobs: - test_endpoints: - runs-on: ubuntu-latest - timeout-minutes: 30 - env: - PRIVATE_KEY: ${{ secrets.PRIVATE_KEY }} - X402_WALLET_ADDRESS: ${{ secrets.X402_WALLET_ADDRESS }} - FUNDING_FAUCET_URL: ${{ secrets.FUNDING_FAUCET_URL }} - BLUESKY_DID: ${{ secrets.BLUESKY_DID }} - BLUESKY_APP_PASSWORD: ${{ secrets.BLUESKY_APP_PASSWORD }} - AP_ACCESS_TOKEN: ${{ secrets.AP_ACCESS_TOKEN }} - TWITTER_BEARER_TOKEN: ${{ secrets.TWITTER_BEARER_TOKEN }} - CDP_API_KEY_ID: ${{ secrets.CDP_API_KEY_ID }} - CDP_API_SECRET: ${{ secrets.CDP_API_SECRET }} - - steps: - - name: Checkout Code - uses: actions/checkout@v4 - - - name: Load model config - run: | - source config/models.env - echo "EMBEDDING_MODEL=$EMBEDDING_MODEL" >> $FORGEJO_ENV - echo "STYLE_MODEL=$STYLE_MODEL" >> $FORGEJO_ENV - - - name: Set up Go - uses: actions/setup-go@v5 - with: - go-version: ${{ env.GO_VERSION }} - cache: true - - - name: Install Dependencies - run: | - go mod download - # Install foundry (cast) for wallet funding - curl -L https://foundry.paradigm.xyz | bash - echo "$HOME/.foundry/bin" >> $FORGEJO_PATH - $HOME/.foundry/bin/foundryup - # Install bc for float math - sudo apt-get update && sudo apt-get install -y bc - - - name: Verify Code Formatting - run: | - if [ -n "$(gofmt -l .)" ]; then - echo "Code is not formatted correctly. Run 'gofmt -w .'" - exit 1 - fi - - - name: Start Test Infrastructure - run: | - docker-compose -f docker-compose.test.yml up -d - - - name: Wait for PostgreSQL - run: | - for i in {1..30}; do - if docker-compose -f docker-compose.test.yml exec -T postgres pg_isready -U postgres -d mcpserver >/dev/null 2>&1; then - echo "PostgreSQL is ready" - exit 0 - fi - sleep 2 - done - echo "PostgreSQL failed to start" - exit 1 - - - name: Wait for Ollama - run: | - for i in {1..30}; do - if curl -sf http://localhost:11434/api/tags >/dev/null 2>&1; then - echo "Ollama is ready" - exit 0 - fi - sleep 2 - done - echo "Ollama failed to start" - exit 1 - - - name: Pull Ollama Models - run: | - docker-compose -f docker-compose.test.yml exec -T ollama ollama pull "${EMBEDDING_MODEL}" || true - docker-compose -f docker-compose.test.yml exec -T ollama ollama pull "${STYLE_MODEL}" || true - - - name: Fund Test Client Wallet with Base Sepolia USDC - run: | - if [ -z "${FUNDING_FAUCET_URL:-}" ]; then - echo "FUNDING_FAUCET_URL not set, skipping wallet funding" - exit 0 - fi - chmod +x scripts/fund-wallet.sh - ./scripts/fund-wallet.sh - - - name: Fund Server Wallet with Base Sepolia USDC - run: | - if [ -z "${FUNDING_FAUCET_URL:-}" ]; then - echo "FUNDING_FAUCET_URL not set, skipping server wallet funding" - exit 0 - fi - # Fund the server wallet (X402_WALLET_ADDRESS) using the same faucet - # This uses the PRIVATE_KEY from env which corresponds to the test client - # For server wallet, we'd need a separate private key, but for testing we use test client key - echo "Server wallet funding uses same faucet (test client wallet)" - - - name: Start Test Server - run: | - chmod +x scripts/start-test-server.sh - ./scripts/start-test-server.sh - - - name: Build Test Client - run: | - go build -o test-endpoints ./testClient - - - name: Run Endpoint Tests - env: - RPC_URL: ${{ env.RPC_URL }} - FUNDING_FAUCET_URL: ${{ secrets.FUNDING_FAUCET_URL }} - run: | - ./test-endpoints - - - name: Cleanup - if: always() - run: | - if [ -f /tmp/test-server.pid ]; then - kill $(cat /tmp/test-server.pid) 2>/dev/null || true - fi - docker-compose -f docker-compose.test.yml down -v \ No newline at end of file diff --git a/.gitignore b/.gitignore index c678aa0..3476089 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,3 @@ .env cdpScripts/cdp -bin/ -test-endpoints -test-endpoints.exe \ No newline at end of file +bin/ \ No newline at end of file diff --git a/cmd/server/main.go b/cmd/server/main.go index a35f99c..a1b31db 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, cfg.EmbeddingModel, cfg.StyleModel) + ollamaClient := ollama.New(ctx, uri, http.DefaultClient) apClient, err := activitypub.NewAPClient(ctx, cfg.ActivityPubHost, cfg.ActivityPubToken) if err != nil { diff --git a/config/models.env b/config/models.env deleted file mode 100644 index 96a8c53..0000000 --- a/config/models.env +++ /dev/null @@ -1,5 +0,0 @@ -# Ollama model configuration -# Single source of truth for model names used by the application and CI - -EMBEDDING_MODEL=qwen3-embedding:0.6b -STYLE_MODEL=qwen3:4b-instruct-2507-q4_K_M \ No newline at end of file diff --git a/docker-compose.test.yml b/docker-compose.test.yml deleted file mode 100644 index 83f6046..0000000 --- a/docker-compose.test.yml +++ /dev/null @@ -1,26 +0,0 @@ -version: '3.8' - -services: - postgres: - image: pgvector/pgvector:pg16 - environment: - POSTGRES_DB: mcpserver - POSTGRES_USER: postgres - POSTGRES_PASSWORD: postgres - ports: - - "5432:5432" - healthcheck: - test: ["CMD-SHELL", "pg_isready -U postgres -d mcpserver"] - interval: 5s - timeout: 5s - retries: 10 - - ollama: - image: ollama/ollama:latest - ports: - - "11434:11434" - healthcheck: - test: ["CMD", "curl", "-f", "http://localhost:11434/api/tags"] - interval: 10s - timeout: 5s - retries: 10 \ No newline at end of file diff --git a/go.mod b/go.mod index a168fc6..5a95b31 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 + github.com/ethereum/go-ethereum v1.17.2 // indirect 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 824768b..a5d4f0e 100644 --- a/internal/activitypub/types.go +++ b/internal/activitypub/types.go @@ -3,11 +3,10 @@ package activitypub import "time" type Post struct { - ID string `json:"id"` - Author string `json:"author"` - Content string `json:"content"` - Text string `json:"text"` - CreatedAt time.Time `json:"created_at"` + 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"` @@ -15,14 +14,13 @@ type Post struct { func (i *internalPost) toPost() *Post { return &Post{ - 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, + 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 deleted file mode 100644 index 02241a6..0000000 --- a/internal/clustering/helpers.go +++ /dev/null @@ -1,37 +0,0 @@ -package clustering - -import "math" - -func normalize(v []float64) []float64 { - out := make([]float64, len(v)) - - var norm float64 - for _, x := range v { - norm += x * x - } - - norm = math.Sqrt(norm) - - if norm == 0 { - return out - } - - for i, x := range v { - out[i] = x / norm - } - - return out -} - -func cosineSimilarity(a, b []float64) float64 { - if len(a) != len(b) { - return -1 - } - - var dot float64 - for i := range a { - dot += a[i] * b[i] - } - - return dot -} diff --git a/internal/clustering/k-means.go b/internal/clustering/k-means.go deleted file mode 100644 index d1a126b..0000000 --- a/internal/clustering/k-means.go +++ /dev/null @@ -1,228 +0,0 @@ -package clustering - -import ( - "errors" - "math" - "math/rand" -) - -func initializeKMeansPlusPlus( - vectors [][]float64, - k int, - rng *rand.Rand, -) [][]float64 { - centroids := make([][]float64, 0, k) - - first := rng.Intn(len(vectors)) - centroids = append( - centroids, - append([]float64(nil), vectors[first]...), - ) - - distances := make([]float64, len(vectors)) - - for len(centroids) < k { - var total float64 - - for i, vector := range vectors { - bestSimilarity := -math.MaxFloat64 - - for _, centroid := range centroids { - similarity := cosineSimilarity(vector, centroid) - if similarity > bestSimilarity { - bestSimilarity = similarity - } - } - - // cosine distance - distance := 1 - bestSimilarity - - if distance < 0 { - distance = 0 - } - - // k-means++ uses squared distance. - distances[i] = distance * distance - total += distances[i] - } - - if total == 0 { - break - } - - target := rng.Float64() * total - - var cumulative float64 - selected := len(vectors) - 1 - - for i, distance := range distances { - cumulative += distance - - if cumulative >= target { - selected = i - break - } - } - - centroids = append( - centroids, - append([]float64(nil), vectors[selected]...), - ) - } - - // This can theoretically happen if all vectors are identical. - for len(centroids) < k { - index := rng.Intn(len(vectors)) - - centroids = append( - centroids, - append([]float64(nil), vectors[index]...), - ) - } - - return centroids -} - -func KMeans(items []Item, k int, maxIterations int, seed int64) ([]Result, error) { - if len(items) == 0 { - return nil, errors.New("no items") - } - - if k <= 0 { - return nil, errors.New("k must be positive") - } - - if k > len(items) { - k = len(items) - } - - if maxIterations <= 0 { - maxIterations = 50 - } - - dimension := len(items[0].Embedding) - if dimension == 0 { - return nil, errors.New("empty embedding") - } - - // Normalize embeddings once. - normalized := make([][]float64, len(items)) - for i, item := range items { - if len(item.Embedding) != dimension { - return nil, errors.New("embedding dimensions do not match") - } - - normalized[i] = normalize(item.Embedding) - } - - rng := rand.New(rand.NewSource(seed)) - - centroids := initializeKMeansPlusPlus(normalized, k, rng) - - assignments := make([]int, len(items)) - - for iteration := 0; iteration < maxIterations; iteration++ { - changed := false - - // Assignment step. - for i, vector := range normalized { - bestCluster := 0 - bestSimilarity := -math.MaxFloat64 - - for cluster, centroid := range centroids { - similarity := cosineSimilarity(vector, centroid) - - if similarity > bestSimilarity { - bestSimilarity = similarity - bestCluster = cluster - } - } - - if iteration == 0 || assignments[i] != bestCluster { - assignments[i] = bestCluster - changed = true - } - } - - // Stop if nothing changed. - if !changed && iteration > 0 { - break - } - - // Update step. - newCentroids := make([][]float64, k) - counts := make([]int, k) - - for cluster := 0; cluster < k; cluster++ { - newCentroids[cluster] = make([]float64, dimension) - } - - for i, vector := range normalized { - cluster := assignments[i] - counts[cluster]++ - - for d, value := range vector { - newCentroids[cluster][d] += value - } - } - - for cluster := 0; cluster < k; cluster++ { - if counts[cluster] == 0 { - // Empty cluster: reinitialize with a random item. - index := rng.Intn(len(normalized)) - newCentroids[cluster] = append( - []float64(nil), - normalized[index]..., - ) - continue - } - - newCentroids[cluster] = normalize(newCentroids[cluster]) - } - - centroids = newCentroids - } - - results := make([]Result, k) - - for cluster := 0; cluster < k; cluster++ { - results[cluster] = Result{ - Cluster: cluster, - Centroid: centroids[cluster], - } - } - - for i, item := range items { - cluster := assignments[i] - - item.Embedding = normalized[i] - - results[cluster].Items = append( - results[cluster].Items, - item, - ) - } - - // Find the item closest to each centroid. - for cluster := range results { - if len(results[cluster].Items) == 0 { - continue - } - - bestSimilarity := -math.MaxFloat64 - - for _, item := range results[cluster].Items { - similarity := cosineSimilarity( - item.Embedding, - results[cluster].Centroid, - ) - - if similarity > bestSimilarity { - bestSimilarity = similarity - results[cluster].Representative = item - } - } - } - - return results, nil -} diff --git a/internal/clustering/kmeans_test.go b/internal/clustering/kmeans_test.go deleted file mode 100644 index 912fdf0..0000000 --- a/internal/clustering/kmeans_test.go +++ /dev/null @@ -1,273 +0,0 @@ -package clustering - -import ( - "math" - "math/rand" - "testing" -) - -func TestKMeans_Basic(t *testing.T) { - items := []Item{ - {ID: "1", Text: "a", Embedding: []float64{1.0, 0.0}}, - {ID: "2", Text: "b", Embedding: []float64{0.9, 0.1}}, - {ID: "3", Text: "c", Embedding: []float64{-1.0, 0.0}}, - {ID: "4", Text: "d", Embedding: []float64{-0.9, -0.1}}, - } - - results, err := KMeans(items, 2, 50, 42) - if err != nil { - t.Fatalf("KMeans failed: %v", err) - } - - if len(results) != 2 { - t.Fatalf("expected 2 clusters, got %d", len(results)) - } - - // Check that items are assigned to clusters - totalItems := 0 - for _, r := range results { - totalItems += len(r.Items) - } - if totalItems != len(items) { - t.Fatalf("expected %d items across clusters, got %d", len(items), totalItems) - } - - // Verify centroids are normalized - for _, r := range results { - norm := 0.0 - for _, v := range r.Centroid { - norm += v * v - } - norm = math.Sqrt(norm) - if math.Abs(norm-1.0) > 1e-6 { - t.Errorf("centroid not normalized: norm=%f", norm) - } - } -} - -func TestKMeans_Empty(t *testing.T) { - _, err := KMeans([]Item{}, 2, 50, 42) - if err == nil { - t.Fatal("expected error for empty items") - } -} - -func TestKMeans_KGreaterThanItems(t *testing.T) { - items := []Item{ - {ID: "1", Text: "a", Embedding: []float64{1.0, 0.0}}, - {ID: "2", Text: "b", Embedding: []float64{0.0, 1.0}}, - } - - results, err := KMeans(items, 5, 50, 42) - if err != nil { - t.Fatalf("KMeans failed: %v", err) - } - - // k should be capped to len(items) - if len(results) != 2 { - t.Fatalf("expected 2 clusters (capped), got %d", len(results)) - } -} - -func TestKMeans_IdenticalVectors(t *testing.T) { - items := []Item{ - {ID: "1", Text: "a", Embedding: []float64{1.0, 0.0}}, - {ID: "2", Text: "b", Embedding: []float64{1.0, 0.0}}, - {ID: "3", Text: "c", Embedding: []float64{1.0, 0.0}}, - } - - results, err := KMeans(items, 2, 50, 42) - if err != nil { - t.Fatalf("KMeans failed: %v", err) - } - - if len(results) != 2 { - t.Fatalf("expected 2 clusters, got %d", len(results)) - } -} - -func TestKMeans_SingleItem(t *testing.T) { - items := []Item{ - {ID: "1", Text: "a", Embedding: []float64{1.0, 0.0}}, - } - - results, err := KMeans(items, 1, 50, 42) - if err != nil { - t.Fatalf("KMeans failed: %v", err) - } - - if len(results) != 1 { - t.Fatalf("expected 1 cluster, got %d", len(results)) - } - if len(results[0].Items) != 1 { - t.Fatalf("expected 1 item in cluster, got %d", len(results[0].Items)) - } -} - -func TestKMeans_DeterministicWithSeed(t *testing.T) { - items := []Item{ - {ID: "1", Text: "a", Embedding: []float64{1.0, 0.0}}, - {ID: "2", Text: "b", Embedding: []float64{0.9, 0.1}}, - {ID: "3", Text: "c", Embedding: []float64{-1.0, 0.0}}, - {ID: "4", Text: "d", Embedding: []float64{-0.9, -0.1}}, - } - - results1, _ := KMeans(items, 2, 50, 42) - results2, _ := KMeans(items, 2, 50, 42) - - // Same seed should produce same assignments - for i := range results1 { - if len(results1[i].Items) != len(results2[i].Items) { - t.Fatalf("cluster %d size differs: %d vs %d", i, len(results1[i].Items), len(results2[i].Items)) - } - } -} - -func TestKMeans_DifferentSeedsMayDiffer(t *testing.T) { - items := []Item{ - {ID: "1", Text: "a", Embedding: []float64{1.0, 0.0}}, - {ID: "2", Text: "b", Embedding: []float64{0.9, 0.1}}, - {ID: "3", Text: "c", Embedding: []float64{-1.0, 0.0}}, - {ID: "4", Text: "d", Embedding: []float64{-0.9, -0.1}}, - } - - results1, _ := KMeans(items, 2, 50, 42) - results2, _ := KMeans(items, 2, 50, 123) - - // Different seeds may produce different cluster sizes (though often same for well-separated data) - // Just verify both are valid - for _, r := range results1 { - if len(r.Items) == 0 { - t.Fatal("seed 42 produced empty cluster") - } - } - for _, r := range results2 { - if len(r.Items) == 0 { - t.Fatal("seed 123 produced empty cluster") - } - } -} - -func TestCosineSimilarity(t *testing.T) { - tests := []struct { - name string - a, b []float64 - expected float64 - }{ - {"identical", []float64{1, 0}, []float64{1, 0}, 1.0}, - {"orthogonal", []float64{1, 0}, []float64{0, 1}, 0.0}, - {"opposite", []float64{1, 0}, []float64{-1, 0}, -1.0}, - {"scaled", []float64{2, 0}, []float64{1, 0}, 2.0}, // not normalized - {"mismatched len", []float64{1, 0}, []float64{1, 0, 0}, -1.0}, - } - - for _, tc := range tests { - t.Run(tc.name, func(t *testing.T) { - got := cosineSimilarity(tc.a, tc.b) - if math.Abs(got-tc.expected) > 1e-6 { - t.Errorf("cosineSimilarity(%v, %v) = %f, want %f", tc.a, tc.b, got, tc.expected) - } - }) - } -} - -func TestNormalize(t *testing.T) { - v := []float64{3.0, 4.0} - normalized := normalize(v) - - // Check length is 1 - norm := 0.0 - for _, x := range normalized { - norm += x * x - } - norm = math.Sqrt(norm) - if math.Abs(norm-1.0) > 1e-6 { - t.Errorf("normalized vector has norm %f, want 1.0", norm) - } - - // Check zero vector handling - zero := normalize([]float64{0.0, 0.0}) - if zero[0] != 0.0 || zero[1] != 0.0 { - t.Errorf("zero vector normalization failed: %v", zero) - } -} - -func TestInitializeKMeansPlusPlus(t *testing.T) { - vectors := [][]float64{ - {1.0, 0.0}, - {0.9, 0.1}, - {-1.0, 0.0}, - {-0.9, -0.1}, - } - - rng := rand.New(rand.NewSource(42)) - centroids := initializeKMeansPlusPlus(vectors, 2, rng) - - if len(centroids) != 2 { - t.Fatalf("expected 2 centroids, got %d", len(centroids)) - } - - // Centroids should be from the input vectors (or averages) - found1, found2 := false, false - for _, c := range centroids { - for _, v := range vectors { - if equalVectors(c, v) { - if !found1 { - found1 = true - } else { - found2 = true - } - } - } - } - if !found1 || !found2 { - t.Errorf("centroids not from input vectors: %v", centroids) - } -} - -func equalVectors(a, b []float64) bool { - if len(a) != len(b) { - return false - } - for i := range a { - if math.Abs(a[i]-b[i]) > 1e-6 { - return false - } - } - return true -} - -func TestKMeans_HighDimensional(t *testing.T) { - // Test with 1024-dim vectors (like qwen3-embedding) - dim := 1024 - items := make([]Item, 10) - for i := range items { - emb := make([]float64, dim) - if i < 5 { - emb[0] = 1.0 // cluster 1 - } else { - emb[dim-1] = 1.0 // cluster 2 - } - items[i] = Item{ - ID: string(rune('a' + i)), - Text: "test", - Embedding: emb, - } - } - - results, err := KMeans(items, 2, 50, 42) - if err != nil { - t.Fatalf("KMeans failed: %v", err) - } - - if len(results) != 2 { - t.Fatalf("expected 2 clusters, got %d", len(results)) - } - - // Verify each cluster has ~5 items - for _, r := range results { - if len(r.Items) < 3 || len(r.Items) > 7 { - t.Errorf("cluster size %d unexpected", len(r.Items)) - } - } -} \ No newline at end of file diff --git a/internal/clustering/types.go b/internal/clustering/types.go deleted file mode 100644 index d1608d1..0000000 --- a/internal/clustering/types.go +++ /dev/null @@ -1,21 +0,0 @@ -package clustering - -import "time" - -type Item struct { - ID string - Text string - Embedding []float64 - Timestamp time.Time - Likes int64 - Reposts int64 - Replies int64 - Quotes int64 -} - -type Result struct { - Cluster int - Items []Item - Centroid []float64 - Representative Item -} diff --git a/internal/config/config.go b/internal/config/config.go index 219a890..29a3228 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -2,10 +2,8 @@ package config import ( "os" - "path/filepath" "strconv" "strings" - "time" "github.com/joho/godotenv" log "github.com/nacorid/logger" @@ -40,32 +38,9 @@ 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") } @@ -91,7 +66,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", "true"), + MigrateOnStart: getEnvBool("MIGRATE_ON_START", "false"), MigrationsDir: getEnv("MIGRATIONS_DIR", "sql/schema"), ReadyzPort: getEnv("READYZ_PORT", "9090"), MetricsPort: getEnv("METRICS_PORT", "9091"), @@ -103,13 +78,6 @@ 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 @@ -143,40 +111,12 @@ func getEnvBool(key, fallback string) bool { } } -func getEnvDuration(key, fallback string) time.Duration { - v := getEnv(key, fallback) - d, err := time.ParseDuration(v) - if err != nil { - d, _ = time.ParseDuration(fallback) - } - return d -} - func splitComma(s string) []string { parts := make([]string, 0) - for p := range strings.SplitSeq(s, ",") { + for _, p := range strings.Split(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 69df20c..15db457 100644 --- a/internal/ollama/client.go +++ b/internal/ollama/client.go @@ -6,64 +6,39 @@ import ( "net/url" "strings" - "github.com/nacorid/naco-api/internal/clustering" "github.com/ollama/ollama/api" ) type OllamaClient struct { - client api.Client - embeddingModel string - styleModel string + client api.Client } -func New(ctx context.Context, host *url.URL, http *http.Client, em, sm string) *OllamaClient { +func New(ctx context.Context, host *url.URL, http *http.Client) *OllamaClient { client := api.NewClient(host, http) - return &OllamaClient{client: *client, embeddingModel: em, styleModel: sm} + return &OllamaClient{client: *client} } -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 { +func (o *OllamaClient) EmbedTexts(ctx context.Context, posts []string) ([][]float64, error) { + vectors := make([][]float64, 0, len(posts)) + for _, t := range posts { resp, err := o.client.Embeddings(ctx, &api.EmbeddingRequest{ - Model: o.embeddingModel, - Prompt: post.Text, + Model: "nomic-embed-text", + Prompt: t, }) if err != nil { return nil, err } - embeddings = append(embeddings, resp.Embedding) + vectors = append(vectors, resp.Embedding) } - - return embeddings, nil + return vectors, nil } -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:" - +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:" var fullResponse strings.Builder err := o.client.Generate(ctx, &api.GenerateRequest{ - Model: o.styleModel, + Model: "mistral", 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 deleted file mode 100644 index 17111f7..0000000 --- a/internal/ollama/client_test.go +++ /dev/null @@ -1,260 +0,0 @@ -package ollama - -import ( - "context" - "net/http" - "net/http/httptest" - "net/url" - "testing" - - "github.com/nacorid/naco-api/internal/clustering" -) - -func TestEmbedTexts(t *testing.T) { - mux := http.NewServeMux() - mux.HandleFunc("/api/embeddings", func(w http.ResponseWriter, r *http.Request) { - w.Header().Set("Content-Type", "application/json") - w.Write([]byte(`{"embedding":[0.1,0.2,0.3]}`)) - }) - - server := httptest.NewServer(mux) - defer server.Close() - - host, _ := url.Parse(server.URL) - client := New(context.Background(), host, http.DefaultClient, "test-embed", "test-style") - - posts := []clustering.Item{ - {ID: "1", Text: "hello", Embedding: nil}, - {ID: "2", Text: "world", Embedding: nil}, - } - - embeddings, err := client.EmbedTexts(context.Background(), posts) - if err != nil { - t.Fatalf("EmbedTexts failed: %v", err) - } - - if len(embeddings) != 2 { - t.Fatalf("expected 2 embeddings, got %d", len(embeddings)) - } - - for i, emb := range embeddings { - if len(emb) != 3 { - t.Errorf("embedding %d has wrong dim: %d", i, len(emb)) - } - if emb[0] != 0.1 || emb[1] != 0.2 || emb[2] != 0.3 { - t.Errorf("embedding %d wrong values: %v", i, emb) - } - } -} - -func TestEmbedTexts_Error(t *testing.T) { - mux := http.NewServeMux() - mux.HandleFunc("/api/embeddings", func(w http.ResponseWriter, r *http.Request) { - w.WriteHeader(http.StatusInternalServerError) - w.Write([]byte("server error")) - }) - - server := httptest.NewServer(mux) - defer server.Close() - - host, _ := url.Parse(server.URL) - client := New(context.Background(), host, http.DefaultClient, "test-embed", "test-style") - - posts := []clustering.Item{{ID: "1", Text: "hello", Embedding: nil}} - - _, err := client.EmbedTexts(context.Background(), posts) - if err == nil { - t.Fatal("expected error from failing server") - } -} - -func TestSummarizeStyle(t *testing.T) { - mux := http.NewServeMux() - mux.HandleFunc("/api/generate", func(w http.ResponseWriter, r *http.Request) { - flusher, ok := w.(http.Flusher) - if !ok { - t.Fatal("response writer doesn't support flushing") - } - w.Header().Set("Content-Type", "application/x-ndjson") - // Stream response as newline-delimited JSON - w.Write([]byte(`{"response":"Style summary: "}` + "\n")) - if flusher != nil { - flusher.Flush() - } - w.Write([]byte(`{"response":"concise and clear."}` + "\n")) - }) - - server := httptest.NewServer(mux) - defer server.Close() - - host, _ := url.Parse(server.URL) - client := New(context.Background(), host, http.DefaultClient, "test-embed", "test-style") - - posts := []clustering.Item{ - {ID: "1", Text: "First post", Embedding: nil}, - {ID: "2", Text: "Second post", Embedding: nil}, - } - - summary, err := client.SummarizeStyle(context.Background(), posts) - if err != nil { - t.Fatalf("SummarizeStyle failed: %v", err) - } - - if summary != "Style summary: concise and clear." { - t.Errorf("unexpected summary: %q", summary) - } -} - -func TestSummarizeStyle_Error(t *testing.T) { - mux := http.NewServeMux() - mux.HandleFunc("/api/generate", func(w http.ResponseWriter, r *http.Request) { - w.WriteHeader(http.StatusInternalServerError) - w.Write([]byte("server error")) - }) - - server := httptest.NewServer(mux) - defer server.Close() - - host, _ := url.Parse(server.URL) - client := New(context.Background(), host, http.DefaultClient, "test-embed", "test-style") - - posts := []clustering.Item{{ID: "1", Text: "hello", Embedding: nil}} - - _, err := client.SummarizeStyle(context.Background(), posts) - if err == nil { - t.Fatal("expected error from failing server") - } -} - -func TestSummarizeStyle_PromptStructure(t *testing.T) { - var receivedPrompt string - mux := http.NewServeMux() - mux.HandleFunc("/api/generate", func(w http.ResponseWriter, r *http.Request) { - // Capture the prompt from request body - receivedPrompt = "captured" - w.Header().Set("Content-Type", "application/json") - w.Write([]byte(`{"response":"ok"}`)) - }) - - server := httptest.NewServer(mux) - defer server.Close() - - host, _ := url.Parse(server.URL) - client := New(context.Background(), host, http.DefaultClient, "test-embed", "test-style") - - posts := []clustering.Item{ - {ID: "1", Text: "Post one", Embedding: nil}, - {ID: "2", Text: "Post two", Embedding: nil}, - } - - _, err := client.SummarizeStyle(context.Background(), posts) - if err != nil { - t.Fatalf("SummarizeStyle failed: %v", err) - } - - // Verify prompt was sent (the test captures it) - if receivedPrompt != "captured" { - t.Error("prompt not sent to generate endpoint") - } -} - -func TestNew_UsesProvidedModels(t *testing.T) { - host, _ := url.Parse("http://localhost:11434") - client := New(context.Background(), host, http.DefaultClient, "custom-embed", "custom-style") - - if client.embeddingModel != "custom-embed" { - t.Errorf("embeddingModel = %q, want %q", client.embeddingModel, "custom-embed") - } - if client.styleModel != "custom-style" { - t.Errorf("styleModel = %q, want %q", client.styleModel, "custom-style") - } -} - -// Test with actual api.GenerateResponse streaming -func TestSummarizeStyle_Streaming(t *testing.T) { - callCount := 0 - mux := http.NewServeMux() - mux.HandleFunc("/api/generate", func(w http.ResponseWriter, r *http.Request) { - callCount++ - flusher, ok := w.(http.Flusher) - if !ok { - t.Fatal("response writer doesn't support flushing") - } - w.Header().Set("Content-Type", "application/x-ndjson") - // Simulate streaming response - w.Write([]byte(`{"response":"part1"}` + "\n")) - if flusher != nil { - flusher.Flush() - } - w.Write([]byte(`{"response":"part2"}` + "\n")) - if flusher != nil { - flusher.Flush() - } - w.Write([]byte(`{"response":"part3","done":true}` + "\n")) - }) - - server := httptest.NewServer(mux) - defer server.Close() - - host, _ := url.Parse(server.URL) - client := New(context.Background(), host, http.DefaultClient, "test-embed", "test-style") - - posts := []clustering.Item{{ID: "1", Text: "test", Embedding: nil}} - - summary, err := client.SummarizeStyle(context.Background(), posts) - if err != nil { - t.Fatalf("SummarizeStyle failed: %v", err) - } - - if summary != "part1part2part3" { - t.Errorf("streaming summary = %q, want %q", summary, "part1part2part3") - } - - if callCount != 1 { - t.Errorf("generate endpoint called %d times, want 1", callCount) - } -} - -func TestEmbedTexts_EmptyInput(t *testing.T) { - mux := http.NewServeMux() - mux.HandleFunc("/api/embeddings", func(w http.ResponseWriter, r *http.Request) { - t.Fatal("embeddings endpoint should not be called for empty input") - }) - - server := httptest.NewServer(mux) - defer server.Close() - - host, _ := url.Parse(server.URL) - client := New(context.Background(), host, http.DefaultClient, "test-embed", "test-style") - - embeddings, err := client.EmbedTexts(context.Background(), []clustering.Item{}) - if err != nil { - t.Fatalf("EmbedTexts with empty input failed: %v", err) - } - - if len(embeddings) != 0 { - t.Errorf("expected empty embeddings, got %d", len(embeddings)) - } -} - -func TestSummarizeStyle_EmptyInput(t *testing.T) { - mux := http.NewServeMux() - mux.HandleFunc("/api/generate", func(w http.ResponseWriter, r *http.Request) { - t.Fatal("generate endpoint should not be called for empty input") - }) - - server := httptest.NewServer(mux) - defer server.Close() - - host, _ := url.Parse(server.URL) - client := New(context.Background(), host, http.DefaultClient, "test-embed", "test-style") - - summary, err := client.SummarizeStyle(context.Background(), []clustering.Item{}) - if err != nil { - t.Fatalf("SummarizeStyle with empty input failed: %v", err) - } - - if summary != "" { - t.Errorf("expected empty summary, got %q", summary) - } -} \ No newline at end of file diff --git a/internal/server/analytics.go b/internal/server/analytics.go deleted file mode 100644 index 7821306..0000000 --- a/internal/server/analytics.go +++ /dev/null @@ -1,883 +0,0 @@ -package server - -import ( - "context" - "crypto/sha256" - "database/sql" - "encoding/hex" - "encoding/json" - "fmt" - "math" - "sort" - "strings" - "time" - - "github.com/bluesky-social/indigo/api/bsky" - "github.com/modelcontextprotocol/go-sdk/mcp" - log "github.com/nacorid/logger" - "github.com/nacorid/naco-api/internal/clustering" - "github.com/nacorid/naco-api/internal/store" - "github.com/nacorid/naco-api/internal/utils" -) - -const ( - defaultAnalyticsLimit = 50 - maxAnalyticsLimit = 200 - defaultAnalyticsDays = 30 - maxTopicCount = 10 - minTopicCount = 3 - sentimentBucketDuration = 24 * time.Hour -) - -// ---- Analytics request/response helpers ---- - -func parseAnalyticsTimeRange(startStr, endStr string) (time.Time, time.Time, error) { - now := time.Now().UTC() - end := now - start := now.AddDate(0, 0, -defaultAnalyticsDays) - - if endStr != "" { - if t, err := time.Parse(time.RFC3339, endStr); err == nil { - end = t - } - } - if startStr != "" { - if t, err := time.Parse(time.RFC3339, startStr); err == nil { - start = t - } - } - if start.After(end) { - start, end = end, start - } - return start, end, nil -} - -func hashTimeRange(start, end time.Time, limit int) string { - data := fmt.Sprintf("%s|%s|%d", start.Format(time.RFC3339), end.Format(time.RFC3339), limit) - h := sha256.Sum256([]byte(data)) - return hex.EncodeToString(h[:]) -} - -func (s *Server) getAnalyticsCache(ctx context.Context, author, platform, toolName, timeRangeHash string) (json.RawMessage, bool) { - result, err := s.db.GetAnalyticsCache(ctx, store.GetAnalyticsCacheParams{ - UserID: author, - Platform: platform, - ToolName: toolName, - TimeRangeHash: timeRangeHash, - }) - if err != nil { - return nil, false - } - return result, true -} - -func (s *Server) setAnalyticsCache(ctx context.Context, author, platform, toolName, timeRangeHash string, result any, ttl time.Duration) error { - data, err := json.Marshal(result) - if err != nil { - return err - } - expiresAt := time.Now().Add(ttl) - return s.db.UpsertAnalyticsCache(ctx, store.UpsertAnalyticsCacheParams{ - UserID: author, - Platform: platform, - ToolName: toolName, - TimeRangeHash: timeRangeHash, - Result: data, - ExpiresAt: expiresAt, - }) -} - -// ---- Main analytics entry points ---- - -func (s *Server) getEngagementMetrics(ctx context.Context, request *mcp.CallToolRequest) (*mcp.CallToolResult, error) { - var args AnalyticsRequest - if err := json.Unmarshal(request.Params.Arguments, &args); err != nil { - return nil, fmt.Errorf("invalid arguments: %w", err) - } - return s.computeAnalytics(ctx, args, "engagementMetrics", s.computeEngagementMetrics) -} - -func (s *Server) getPostingPatterns(ctx context.Context, request *mcp.CallToolRequest) (*mcp.CallToolResult, error) { - var args AnalyticsRequest - if err := json.Unmarshal(request.Params.Arguments, &args); err != nil { - return nil, fmt.Errorf("invalid arguments: %w", err) - } - return s.computeAnalytics(ctx, args, "postingPatterns", s.computePostingPatterns) -} - -func (s *Server) getTopicModeling(ctx context.Context, request *mcp.CallToolRequest) (*mcp.CallToolResult, error) { - var args AnalyticsRequest - if err := json.Unmarshal(request.Params.Arguments, &args); err != nil { - return nil, fmt.Errorf("invalid arguments: %w", err) - } - return s.computeAnalytics(ctx, args, "topicModeling", s.computeTopicModeling) -} - -func (s *Server) getSentimentTrends(ctx context.Context, request *mcp.CallToolRequest) (*mcp.CallToolResult, error) { - var args AnalyticsRequest - if err := json.Unmarshal(request.Params.Arguments, &args); err != nil { - return nil, fmt.Errorf("invalid arguments: %w", err) - } - return s.computeAnalytics(ctx, args, "sentimentTrends", s.computeSentimentTrends) -} - -func (s *Server) getEngagementDrivers(ctx context.Context, request *mcp.CallToolRequest) (*mcp.CallToolResult, error) { - var args AnalyticsRequest - if err := json.Unmarshal(request.Params.Arguments, &args); err != nil { - return nil, fmt.Errorf("invalid arguments: %w", err) - } - return s.computeAnalytics(ctx, args, "engagementDrivers", s.computeEngagementDrivers) -} - -type analyticsComputer func(ctx context.Context, args AnalyticsRequest, posts []clustering.Item) (any, error) - -func (s *Server) computeAnalytics(ctx context.Context, args AnalyticsRequest, toolName string, compute analyticsComputer) (*mcp.CallToolResult, error) { - // Normalize args - limit := args.Limit - if limit <= 0 { - limit = defaultAnalyticsLimit - } - if limit > maxAnalyticsLimit { - limit = maxAnalyticsLimit - } - - start, end, err := parseAnalyticsTimeRange(args.StartTime, args.EndTime) - if err != nil { - return nil, fmt.Errorf("invalid time range: %w", err) - } - - platforms := []string{} - if args.Platform != "" { - platforms = []string{args.Platform} - } else { - platforms = []string{"Bluesky", "ActivityPub", "Nostr"} - } - - timeRangeHash := hashTimeRange(start, end, limit) - - // The analytics result cache is keyed by the scope of the computation: a - // single platform when one is requested, or the "all" sentinel when the - // result aggregates across all platforms. This keeps aggregate results from - // being mixed with single-platform results. - cachePlatform := "all" - if args.Platform != "" { - cachePlatform = args.Platform - } - - if cached, ok := s.getAnalyticsCache(ctx, args.Author, cachePlatform, toolName, timeRangeHash); ok { - var cachedResult json.RawMessage - json.Unmarshal(cached, &cachedResult) - return s.toToolResult(cachedResult) - } - - // Fetch posts from all platforms - var allPosts []clustering.Item - for _, p := range platforms { - plat, ok := platformFromString(p) - if !ok { - continue - } - posts, err := s.fetchAnalyticsPosts(ctx, plat, args.Author, start, end, limit) - if err != nil { - log.WarnWithContext(ctx, "failed to fetch posts for analytics", "platform", p, "error", err) - continue - } - allPosts = append(allPosts, posts...) - } - - if len(allPosts) == 0 { - return nil, fmt.Errorf("no posts found for author %s in time range", args.Author) - } - - // Compute - result, err := compute(ctx, args, allPosts) - if err != nil { - return nil, err - } - - // Cache the result under the matching scope key. - ttl := s.getCacheTTL(toolName) - s.setAnalyticsCache(ctx, args.Author, cachePlatform, toolName, timeRangeHash, result, ttl) - - return s.toToolResult(result) -} - -func (s *Server) fetchAnalyticsPosts(ctx context.Context, platform platform, author string, start, end time.Time, limit int) ([]clustering.Item, error) { - posts, err := s.fetchPostsInTimeForAnalytics(ctx, platform, author, start, end, limit) - if err != nil { - return nil, err - } - return posts, nil -} - -func (s *Server) getCachedAnalyticsPosts(ctx context.Context, author, platform string, start, end time.Time, limit int) ([]clustering.Item, error) { - rows, err := s.db.GetAnalyticsPostsInRange(ctx, store.GetAnalyticsPostsInRangeParams{ - UserID: author, - Platform: platform, - PostTimestamp: start, - PostTimestamp_2: end, - Limit: int32(limit), - }) - if err != nil { - return nil, err - } - items := make([]clustering.Item, 0, len(rows)) - for _, r := range rows { - items = append(items, clustering.Item{ - ID: r.PostID, - Text: r.PostText, - Timestamp: r.PostTimestamp, - Likes: r.Likes, - Reposts: r.Reposts, - Replies: r.Replies, - Quotes: r.Quotes, - }) - } - return items, nil -} - -func (s *Server) persistAnalyticsPosts(ctx context.Context, author, platform string, posts []clustering.Item) error { - for _, p := range posts { - if p.ID == "" { - continue - } - if err := s.db.UpsertAnalyticsPosts(ctx, store.UpsertAnalyticsPostsParams{ - UserID: author, - Platform: platform, - PostID: p.ID, - PostText: p.Text, - PostTimestamp: p.Timestamp, - Likes: p.Likes, - Reposts: p.Reposts, - Replies: p.Replies, - Quotes: p.Quotes, - }); err != nil { - return err - } - } - return nil -} - -func (s *Server) fetchPostsInTimeForAnalytics(ctx context.Context, platform platform, author string, start, end time.Time, limit int) ([]clustering.Item, error) { - var items []clustering.Item - switch platform { - case Bluesky: - res, err := s.bskyClient.GetPostsInTimeframe(ctx, author, start, end, limit) - if err != nil { - return nil, err - } - for _, p := range res { - if p.Post == nil || p.Post.Record == nil { - continue - } - if fp, ok := p.Post.Record.Val.(*bsky.FeedPost); ok { - item := clustering.Item{ - ID: p.Post.Uri, - Text: fp.Text, - } - if ts, err := time.Parse(time.RFC3339, p.Post.IndexedAt); err == nil { - item.Timestamp = ts - } - item.Likes = int64Val(p.Post.LikeCount) - item.Reposts = int64Val(p.Post.RepostCount) - item.Replies = int64Val(p.Post.ReplyCount) - item.Quotes = int64Val(p.Post.QuoteCount) - items = append(items, item) - } - } - case ActivityPub: - res, err := s.apClient.GetPostsInTimeframe(ctx, author, start, end, limit) - if err != nil { - return nil, err - } - for _, p := range res { - if p.Text != "" { - items = append(items, clustering.Item{ - ID: p.ID, - Text: p.Text, - Timestamp: p.CreatedAt, - Likes: int64(p.FavouritesCount), - Reposts: int64(p.ReblogsCount), - Replies: int64(p.RepliesCount), - }) - } - } - case Nostr: - res, err := s.nostrClient.GetPostsInTimeframe(ctx, author, start, end, limit) - if err != nil { - return nil, err - } - for _, p := range res { - if p.Content != "" { - items = append(items, clustering.Item{ - ID: p.ID, - Text: p.Content, - Timestamp: p.CreatedAt, - }) - } - } - case Twitter: - // Twitter API calls are expensive (paid). Reuse persisted post data and - // only call the API for time ranges that are not yet covered by the - // cache, using a coverage watermark to guarantee broad coverage. - fetched, err := s.fetchTwitterAnalyticsPosts(ctx, author, start, end, limit) - if err != nil { - return nil, err - } - items = append(items, fetched...) - } - return items, nil -} - -// fetchTwitterAnalyticsPosts returns posts in [start, end] for the author, -// fetching from the Twitter API only the sub-ranges not already covered by the -// persistent post cache, and storing any newly fetched posts for future use. -func (s *Server) fetchTwitterAnalyticsPosts(ctx context.Context, author string, start, end time.Time, limit int) ([]clustering.Item, error) { - coverage, err := s.getAnalyticsCoverage(ctx, author, "Twitter") - if err != nil { - // Unknown coverage (missing row) is not an error; fall back to fetching. - if err != sql.ErrNoRows { - return nil, err - } - } - - coverageStart, coverageEnd := time.Time{}, time.Time{} - haveCoverage := !coverage.CoveredStart.IsZero() && !coverage.CoveredEnd.IsZero() - if haveCoverage { - coverageStart, coverageEnd = coverage.CoveredStart, coverage.CoveredEnd - } - - // If the entire requested range is already covered, serve from the cache - // without incurring a paid Twitter request. - if haveCoverage && !start.Before(coverageStart) && !end.After(coverageEnd) { - return s.getCachedAnalyticsPosts(ctx, author, "Twitter", start, end, limit) - } - - // Determine the sub-ranges of [start, end] not yet covered, and fetch each - // from the API. Fetching per uncovered range keeps costs bounded while - // expanding coverage on every call. - ranges := uncoveredRanges(start, end, coverageStart, coverageEnd, haveCoverage) - - var fetched []clustering.Item - for _, r := range ranges { - posts, err := s.fetchTwitterFromAPI(ctx, author, r.start, r.end, limit) - if err != nil { - return nil, err - } - fetched = append(fetched, posts...) - } - - // Persist newly fetched posts permanently so repeat calls never re-pay. - if len(fetched) > 0 { - if err := s.persistAnalyticsPosts(ctx, author, "Twitter", fetched); err != nil { - log.WarnWithContext(ctx, "failed to persist analytics posts", "platform", "Twitter", "error", err) - } - } - - // Extend coverage to encompass the whole requested range. - if err := s.extendAnalyticsCoverage(ctx, author, "Twitter", start, end); err != nil { - log.WarnWithContext(ctx, "failed to extend analytics coverage", "platform", "Twitter", "error", err) - } - - // Return the full set: posts from cache plus anything newly fetched, merged - // and de-duplicated by post ID. - return s.mergeAnalyticsPosts(ctx, author, start, end, limit, fetched) -} - -type timeRange struct { - start, end time.Time -} - -// uncoveredRanges returns the sub-ranges of [start, end] not covered by -// [coveredStart, coveredEnd]. If there is no coverage, the whole range is -// returned. -func uncoveredRanges(start, end, coveredStart, coveredEnd time.Time, have bool) []timeRange { - if !have { - return []timeRange{{start: start, end: end}} - } - - var ranges []timeRange - if start.Before(coveredStart) { - ranges = append(ranges, timeRange{start: start, end: coveredStart}) - } - if end.After(coveredEnd) { - ranges = append(ranges, timeRange{start: coveredEnd, end: end}) - } - return ranges -} - -func (s *Server) getAnalyticsCoverage(ctx context.Context, author, platform string) (store.GetAnalyticsCoverageRow, error) { - return s.db.GetAnalyticsCoverage(ctx, store.GetAnalyticsCoverageParams{ - UserID: author, - Platform: platform, - }) -} - -func (s *Server) extendAnalyticsCoverage(ctx context.Context, author, platform string, start, end time.Time) error { - return s.db.UpsertAnalyticsCoverage(ctx, store.UpsertAnalyticsCoverageParams{ - UserID: author, - Platform: platform, - CoveredStart: start, - CoveredEnd: end, - }) -} - -func (s *Server) fetchTwitterFromAPI(ctx context.Context, author string, start, end time.Time, limit int) ([]clustering.Item, error) { - res, err := s.twitterClient.GetPostsInTimeframe(ctx, author, start, end, limit) - if err != nil { - return nil, err - } - var items []clustering.Item - for _, p := range res { - if p.Content == "" { - continue - } - items = append(items, clustering.Item{ - ID: p.ID, - Text: p.Content, - Timestamp: p.CreatedAt, - Likes: int64(p.Likes), - Reposts: int64(p.Reposts), - Replies: int64(p.Replies), - Quotes: int64(p.Quotes), - }) - } - return items, nil -} - -// mergeAnalyticsPosts combines cached posts in [start, end] with newly fetched -// posts, de-duplicating by post ID. -func (s *Server) mergeAnalyticsPosts(ctx context.Context, author string, start, end time.Time, limit int, fetched []clustering.Item) ([]clustering.Item, error) { - merged := make([]clustering.Item, 0, len(fetched)) - seen := make(map[string]struct{}, len(fetched)) - for _, p := range fetched { - if _, ok := seen[p.ID]; ok { - continue - } - seen[p.ID] = struct{}{} - merged = append(merged, p) - } - - cached, err := s.getCachedAnalyticsPosts(ctx, author, "Twitter", start, end, limit) - if err != nil { - return merged, nil - } - for _, p := range cached { - if _, ok := seen[p.ID]; ok { - continue - } - seen[p.ID] = struct{}{} - merged = append(merged, p) - } - return merged, nil -} - -func (s *Server) toToolResult(result any) (*mcp.CallToolResult, error) { - data, err := json.Marshal(result) - if err != nil { - return nil, err - } - return &mcp.CallToolResult{ - Content: []mcp.Content{ - &mcp.TextContent{Text: string(data)}, - }, - }, nil -} - -func (s *Server) getCacheTTL(toolName string) time.Duration { - switch toolName { - case "postingPatterns": - return s.analyticsTTLPatterns - case "topicModeling": - return s.analyticsTTLTopics - case "sentimentTrends": - return s.analyticsTTLSentiment - case "engagementDrivers": - return s.analyticsTTLDrivers - case "engagementMetrics": - fallthrough - default: - return s.analyticsTTLEngagement - } -} - -// ---- Computation implementations ---- - -func (s *Server) computeEngagementMetrics(ctx context.Context, args AnalyticsRequest, posts []clustering.Item) (any, error) { - var metrics []EngagementMetricPoint - for _, post := range posts { - total := post.Likes + post.Reposts + post.Replies + post.Quotes - // Rate is a simple relative measure (interactions per post). Without a - // follower/fan base to normalize against, report total interactions. - metrics = append(metrics, EngagementMetricPoint{ - PostID: post.ID, - Timestamp: post.Timestamp.Format(time.RFC3339), - Likes: post.Likes, - Reposts: post.Reposts, - Replies: post.Replies, - Quotes: post.Quotes, - EngagementRate: float64(total), - }) - } - - return EngagementMetricsResponse{ - Author: args.Author, - Metrics: metrics, - }, nil -} - -func (s *Server) computePostingPatterns(ctx context.Context, args AnalyticsRequest, posts []clustering.Item) (any, error) { - // Heatmap: 24 hours x 7 days - heatmap := make([][]int, 24) - for i := range heatmap { - heatmap[i] = make([]int, 7) - } - dayCounts := make(map[int]int) - hourCounts := make(map[int]int) - dateSet := make(map[string]bool) - - for _, post := range posts { - t := post.Timestamp - if t.IsZero() { - continue - } - hour := t.Hour() - weekday := int(t.Weekday()) - heatmap[hour][weekday]++ - hourCounts[hour]++ - dayCounts[weekday]++ - dateSet[t.Format("2006-01-02")] = true - } - - // Find peak hours (top 3) - type hourCount struct { - Hour int - Count int - } - var hours []hourCount - for h, c := range hourCounts { - hours = append(hours, hourCount{h, c}) - } - sort.Slice(hours, func(i, j int) bool { return hours[i].Count > hours[j].Count }) - peakHours := make([]int, 0, 3) - for i := 0; i < len(hours) && i < 3; i++ { - peakHours = append(peakHours, hours[i].Hour) - } - - // Find peak days (top 3) - type dayCount struct { - Day int - Count int - } - var days []dayCount - for d, c := range dayCounts { - days = append(days, dayCount{d, c}) - } - sort.Slice(days, func(i, j int) bool { return days[i].Count > days[j].Count }) - peakDays := make([]int, 0, 3) - for i := 0; i < len(days) && i < 3; i++ { - peakDays = append(peakDays, days[i].Day) - } - - totalPosts := len(posts) - postingDays := len(dateSet) - avgPostsPerDay := 0.0 - if postingDays > 0 { - avgPostsPerDay = float64(totalPosts) / float64(postingDays) - } - - return PostingPatternsResponse{ - Author: args.Author, - Heatmap: heatmap, - PeakHours: peakHours, - PeakDays: peakDays, - AvgPostsPerDay: avgPostsPerDay, - TotalPosts: totalPosts, - PostingDays: postingDays, - }, nil -} - -func (s *Server) computeTopicModeling(ctx context.Context, args AnalyticsRequest, posts []clustering.Item) (any, error) { - if len(posts) < minTopicCount { - return TopicModelingResponse{Author: args.Author, Topics: []TopicCluster{}}, nil - } - - // Use embeddings from ollama for clustering - items := posts - // We need embeddings for clustering - compute them - embeddings, err := s.ollama.EmbedTexts(ctx, items) - if err != nil { - return nil, fmt.Errorf("failed to compute embeddings: %w", err) - } - - // Update items with embeddings - for i := range items { - items[i].Embedding = embeddings[i] - } - - // Determine optimal number of topics - topicCount := maxTopicCount - if len(items) < maxTopicCount { - topicCount = len(items) - } - if topicCount < minTopicCount { - topicCount = minTopicCount - } - - // Run KMeans - results, err := clustering.KMeans(items, topicCount, 50, 42) - if err != nil { - return nil, fmt.Errorf("kmeans failed: %w", err) - } - - // Generate labels for each cluster using Ollama - var topics []TopicCluster - for _, cluster := range results { - if len(cluster.Items) == 0 { - continue - } - - // Collect sample texts for labeling - var samples []string - for _, item := range cluster.Items { - if len(samples) >= 3 { - break - } - samples = append(samples, item.Text) - } - - // Generate label using Ollama - label := s.generateTopicLabel(ctx, samples) - keywords := s.extractKeywords(cluster.Items) - - topics = append(topics, TopicCluster{ - TopicID: cluster.Cluster, - Label: label, - Keywords: keywords, - PostCount: len(cluster.Items), - Percentage: float64(len(cluster.Items)) / float64(len(items)) * 100, - SamplePosts: samples, - }) - } - - // Sort by post count descending - sort.Slice(topics, func(i, j int) bool { return topics[i].PostCount > topics[j].PostCount }) - - return TopicModelingResponse{ - Author: args.Author, - Topics: topics, - }, nil -} - -func (s *Server) computeSentimentTrends(ctx context.Context, args AnalyticsRequest, posts []clustering.Item) (any, error) { - // Bucket posts by day - bucketMap := make(map[string][]string) - for _, post := range posts { - if post.Timestamp.IsZero() { - continue - } - bucket := post.Timestamp.Truncate(sentimentBucketDuration).Format(time.RFC3339) - bucketMap[bucket] = append(bucketMap[bucket], post.Text) - } - - var timeSeries []SentimentPoint - overallPos, overallNeg, overallNeu := 0.0, 0.0, 0.0 - emotionCounts := make(map[string]int) - var shifts []SentimentShift - var prevSentiment string - var prevScores [3]float64 - - // Process buckets in chronological order so sentiment shifts are detected correctly. - buckets := make([]string, 0, len(bucketMap)) - for b := range bucketMap { - buckets = append(buckets, b) - } - sort.Strings(buckets) - - for _, bucket := range buckets { - texts := bucketMap[bucket] - pos, neg, neu, emotions := s.analyzeSentiment(ctx, texts) - timeSeries = append(timeSeries, SentimentPoint{ - Timestamp: bucket, - Positive: pos, - Negative: neg, - Neutral: neu, - PostCount: len(texts), - }) - - overallPos += pos * float64(len(texts)) - overallNeg += neg * float64(len(texts)) - overallNeu += neu * float64(len(texts)) - - for e, c := range emotions { - emotionCounts[e] += c - } - - // Detect sentiment shifts - dominant := maxSentiment(pos, neg, neu) - if prevSentiment != "" && dominant != prevSentiment { - mag := math.Abs(pos-prevScores[0]) + math.Abs(neg-prevScores[1]) + math.Abs(neu-prevScores[2]) - shifts = append(shifts, SentimentShift{ - Timestamp: bucket, - FromSentiment: prevSentiment, - ToSentiment: dominant, - Magnitude: mag, - }) - } - prevSentiment = dominant - prevScores = [3]float64{pos, neg, neu} - } - - totalPosts := float64(len(posts)) - if totalPosts > 0 { - overallPos /= totalPosts - overallNeg /= totalPosts - overallNeu /= totalPosts - } - - // Top emotions - var topEmotions []EmotionScore - type emoCount struct { - Emotion string - Count int - } - var emos []emoCount - for e, c := range emotionCounts { - emos = append(emos, emoCount{e, c}) - } - sort.Slice(emos, func(i, j int) bool { return emos[i].Count > emos[j].Count }) - for i := 0; i < len(emos) && i < 5; i++ { - topEmotions = append(topEmotions, EmotionScore{ - Emotion: emos[i].Emotion, - Score: float64(emos[i].Count) / totalPosts, - }) - } - - return SentimentTrendsResponse{ - Author: args.Author, - TimeSeries: timeSeries, - OverallSentiment: SentimentSummary{ - Positive: overallPos, - Negative: overallNeg, - Neutral: overallNeu, - }, - DominantEmotions: topEmotions, - SentimentShifts: shifts, - }, nil -} - -func (s *Server) computeEngagementDrivers(ctx context.Context, args AnalyticsRequest, posts []clustering.Item) (any, error) { - // Topic modeling first - topicResult, err := s.computeTopicModeling(ctx, args, posts) - if err != nil { - return nil, err - } - topicResp := topicResult.(TopicModelingResponse) - - var drivers []EngagementDriver - for _, topic := range topicResp.Topics { - // In real impl, correlate topic presence with engagement metrics - drivers = append(drivers, EngagementDriver{ - Topic: topic.Label, - AvgEngagement: 0.0, - Correlation: 0.0, - PostCount: topic.PostCount, - SamplePosts: topic.SamplePosts[:min(2, len(topic.SamplePosts))], - }) - } - - sort.Slice(drivers, func(i, j int) bool { - return math.Abs(drivers[i].Correlation) > math.Abs(drivers[j].Correlation) - }) - - return EngagementDriversResponse{ - Author: args.Author, - Drivers: drivers, - }, nil -} - -// ---- Helper functions ---- - -func (s *Server) generateTopicLabel(ctx context.Context, samples []string) string { - if len(samples) == 0 { - return "General" - } - prompt := fmt.Sprintf(`Given these sample posts, provide a concise 1-3 word topic label: - -Posts: -%s - -Label only (no explanation):`, strings.Join(samples, "\n---\n")) - - resp, err := s.ollama.SummarizeStyle(ctx, []clustering.Item{{Text: prompt}}) - if err != nil { - return "General" - } - return strings.TrimSpace(resp) -} - -func (s *Server) extractKeywords(items []clustering.Item) []string { - var allText []string - for _, item := range items { - allText = append(allText, item.Text) - } - return utils.TopWords(allText, make(map[string]struct{}), 5) -} - -func (s *Server) analyzeSentiment(ctx context.Context, texts []string) (pos, neg, neu float64, emotions map[string]int) { - if len(texts) == 0 { - return 0, 0, 0, nil - } - - // Use Ollama for sentiment analysis - prompt := fmt.Sprintf(`Analyze sentiment and emotions in these texts. Return JSON only: -{ - "positive": 0.0-1.0, - "negative": 0.0-1.0, - "neutral": 0.0-1.0, - "emotions": {"emotion": count} -} - -Texts: -%s`, strings.Join(texts, "\n---\n")) - - resp, err := s.ollama.SummarizeStyle(ctx, []clustering.Item{{Text: prompt}}) - if err != nil { - return 0.33, 0.33, 0.34, nil - } - - var result struct { - Positive float64 `json:"positive"` - Negative float64 `json:"negative"` - Neutral float64 `json:"neutral"` - Emotions map[string]int `json:"emotions"` - } - json.Unmarshal([]byte(resp), &result) - - // Normalize - sum := result.Positive + result.Negative + result.Neutral - if sum > 0 { - result.Positive /= sum - result.Negative /= sum - result.Neutral /= sum - } - - return result.Positive, result.Negative, result.Neutral, result.Emotions -} - -func maxSentiment(pos, neg, neu float64) string { - if pos >= neg && pos >= neu { - return "positive" - } - if neg >= pos && neg >= neu { - return "negative" - } - return "neutral" -} - -func int64Val(p *int64) int64 { - if p == nil { - return 0 - } - return *p -} diff --git a/internal/server/backends.go b/internal/server/backends.go index 3ca49dd..da9c250 100644 --- a/internal/server/backends.go +++ b/internal/server/backends.go @@ -8,7 +8,6 @@ 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" ) @@ -50,6 +49,6 @@ type twitterBackend interface { } type ollamaBackend interface { - EmbedTexts(ctx context.Context, posts []clustering.Item) ([][]float64, error) - SummarizeStyle(ctx context.Context, posts []clustering.Item) (string, error) + EmbedTexts(ctx context.Context, posts []string) ([][]float64, error) + SummarizeStyle(ctx context.Context, posts []string) (string, error) } diff --git a/internal/server/coverage_test.go b/internal/server/coverage_test.go deleted file mode 100644 index 8e465f7..0000000 --- a/internal/server/coverage_test.go +++ /dev/null @@ -1,71 +0,0 @@ -package server - -import ( - "testing" - "time" -) - -func rangeOf(t *testing.T, s, e string) timeRange { - t.Helper() - layout := "2006-01-02 15:04" - start, err := time.Parse(layout, s) - if err != nil { - t.Fatalf("bad start %q: %v", s, err) - } - end, err := time.Parse(layout, e) - if err != nil { - t.Fatalf("bad end %q: %v", e, err) - } - return timeRange{start: start, end: end} -} - -func assertRanges(t *testing.T, got []timeRange, want ...timeRange) { - t.Helper() - if len(got) != len(want) { - t.Fatalf("got %d ranges, want %d: %+v", len(got), len(want), got) - } - for i := range want { - if !got[i].start.Equal(want[i].start) || !got[i].end.Equal(want[i].end) { - t.Fatalf("range %d: got [%v, %v], want [%v, %v]", - i, got[i].start, got[i].end, want[i].start, want[i].end) - } - } -} - -func TestUncoveredRangesNoCoverage(t *testing.T) { - r := rangeOf(t, "2026-01-01 00:00", "2026-01-31 00:00") - got := uncoveredRanges(r.start, r.end, time.Time{}, time.Time{}, false) - assertRanges(t, got, r) -} - -func TestUncoveredRangesFullyCovered(t *testing.T) { - req := rangeOf(t, "2026-01-10 00:00", "2026-01-20 00:00") - cov := rangeOf(t, "2026-01-01 00:00", "2026-01-31 00:00") - got := uncoveredRanges(req.start, req.end, cov.start, cov.end, true) - assertRanges(t, got) -} - -func TestUncoveredRangesLeftOnly(t *testing.T) { - req := rangeOf(t, "2026-01-01 00:00", "2026-01-31 00:00") - cov := rangeOf(t, "2026-01-15 00:00", "2026-01-31 00:00") - want := rangeOf(t, "2026-01-01 00:00", "2026-01-15 00:00") - got := uncoveredRanges(req.start, req.end, cov.start, cov.end, true) - assertRanges(t, got, want) -} - -func TestUncoveredRangesRightOnly(t *testing.T) { - req := rangeOf(t, "2026-01-01 00:00", "2026-01-31 00:00") - cov := rangeOf(t, "2026-01-01 00:00", "2026-01-15 00:00") - want := rangeOf(t, "2026-01-15 00:00", "2026-01-31 00:00") - got := uncoveredRanges(req.start, req.end, cov.start, cov.end, true) - assertRanges(t, got, want) -} - -func TestUncoveredRangesBothSides(t *testing.T) { - req := rangeOf(t, "2026-01-01 00:00", "2026-01-31 00:00") - cov := rangeOf(t, "2026-01-10 00:00", "2026-01-20 00:00") - wantLeft := rangeOf(t, "2026-01-01 00:00", "2026-01-10 00:00") - wantRight := rangeOf(t, "2026-01-20 00:00", "2026-01-31 00:00") - got := uncoveredRanges(req.start, req.end, cov.start, cov.end, true) - assertRanges(t, got, wantLeft, wantRight) -} diff --git a/internal/server/server.go b/internal/server/server.go index 730ef24..e60f037 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -18,7 +18,6 @@ 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" @@ -31,28 +30,23 @@ 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{} - analyticsTTLEngagement time.Duration - analyticsTTLPatterns time.Duration - analyticsTTLTopics time.Duration - analyticsTTLSentiment time.Duration - analyticsTTLDrivers time.Duration + 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{} } func New(ctx context.Context, db *store.Queries, ap *activitypub.APClient, bsky *bluesky.BlueskyClient, n *nostr.Client, t *twitter.Client, o *ollama.OllamaClient, cfg *config.Config) *Server { @@ -116,7 +110,7 @@ func New(ctx context.Context, db *store.Queries, ap *activitypub.APClient, bsky txHash := settleCtx.Result.Transaction platform := "unknown" payload := map[string]any{ - "event": "transaction_success", + "event": "transaction_success", "transaction": map[string]any{ "x402_transaction_hash": txHash, }, @@ -147,28 +141,23 @@ 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), - analyticsTTLEngagement: cfg.AnalyticsTTLEngagement, - analyticsTTLPatterns: cfg.AnalyticsTTLPatterns, - analyticsTTLTopics: cfg.AnalyticsTTLTopics, - analyticsTTLSentiment: cfg.AnalyticsTTLSentiment, - analyticsTTLDrivers: cfg.AnalyticsTTLDrivers, + 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), } s.registerTools() @@ -474,171 +463,6 @@ 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", @@ -1445,22 +1269,6 @@ func (s *Server) buildStyleProfile(ctx context.Context, platform platform, autho return nil, fmt.Errorf("no posts found for author %s", author) } - // 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 @@ -1470,7 +1278,7 @@ func (s *Server) buildStyleProfile(ctx context.Context, platform platform, autho return nil, err } personaEmbedding := utils.AverageVector(embeddings) - topWords := utils.TopWords(postTexts, make(map[string]struct{}), 10) + topWords := utils.TopWords(posts, make(map[string]struct{}), 10) styleSummary, err := s.ollama.SummarizeStyle(ctx, posts) if err != nil { return nil, err @@ -1581,109 +1389,61 @@ 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) ([]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 +func (s *Server) fetchProfilePosts(ctx context.Context, platform platform, author string, limit int) ([]string, error) { switch platform { case Bluesky: res, err := s.bskyClient.GetLatestNPosts(ctx, author, limit) if err != nil { - return toItems(cached), err + return nil, err } - freshPosts = make([]clustering.Item, 0, len(res)) + posts := make([]string, 0, len(res)) for _, p := range res { fp, ok := p.Post.Record.Val.(*bsky.FeedPost) if !ok { continue } - freshPosts = append(freshPosts, clustering.Item{ - ID: p.Post.Uri, - Text: fp.Text, - }) + posts = append(posts, fp.Text) } + return posts, nil case ActivityPub: res, err := s.apClient.GetLatestNPosts(ctx, author, limit) if err != nil { - return toItems(cached), err + return nil, err } - freshPosts = make([]clustering.Item, 0, len(res)) + posts := make([]string, 0, len(res)) for _, p := range res { if p.Text != "" { - freshPosts = append(freshPosts, clustering.Item{ - ID: p.ID, - Text: p.Text, - }) + posts = append(posts, p.Text) } } + return posts, nil case Nostr: res, err := s.nostrClient.GetLatestNPosts(ctx, author, limit) if err != nil { - return toItems(cached), err + return nil, err } - freshPosts = make([]clustering.Item, 0, len(res)) + posts := make([]string, 0, len(res)) for _, p := range res { if p.Content != "" { - freshPosts = append(freshPosts, clustering.Item{ - ID: p.ID, - Text: p.Content, - }) + posts = append(posts, p.Content) } } + return posts, nil case Twitter: res, err := s.twitterClient.GetLatestNPosts(ctx, author, limit) if err != nil { - return toItems(cached), err + return nil, err } - freshPosts = make([]clustering.Item, 0, len(res)) + posts := make([]string, 0, len(res)) for _, p := range res { if p.Content != "" { - freshPosts = append(freshPosts, clustering.Item{ - ID: p.ID, - Text: p.Content, - }) + posts = append(posts, 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) { @@ -1792,16 +1552,5 @@ 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 d07cd13..57ad7cd 100644 --- a/internal/server/types.go +++ b/internal/server/types.go @@ -151,99 +151,3 @@ 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 75708c1..73b972a 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 len(jsonBytes) == 0 { + if jsonBytes == nil || 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 deleted file mode 100644 index 8ee6cb0..0000000 --- a/internal/store/AnalyticsCache.sql.go +++ /dev/null @@ -1,96 +0,0 @@ -// Code generated by sqlc. DO NOT EDIT. -// versions: -// sqlc v1.31.1 -// source: AnalyticsCache.sql - -package store - -import ( - "context" - "encoding/json" - "time" -) - -const deleteAnalyticsCache = `-- name: DeleteAnalyticsCache :exec -DELETE FROM analytics_cache -WHERE user_id = $1 AND platform = $2 AND tool_name = $3 AND time_range_hash = $4 -` - -type DeleteAnalyticsCacheParams struct { - UserID string - Platform string - ToolName string - TimeRangeHash string -} - -func (q *Queries) DeleteAnalyticsCache(ctx context.Context, arg DeleteAnalyticsCacheParams) error { - _, err := q.db.ExecContext(ctx, deleteAnalyticsCache, - arg.UserID, - arg.Platform, - arg.ToolName, - arg.TimeRangeHash, - ) - return err -} - -const getAnalyticsCache = `-- name: GetAnalyticsCache :one -SELECT result FROM analytics_cache -WHERE user_id = $1 AND platform = $2 AND tool_name = $3 AND time_range_hash = $4 - AND expires_at > now() -` - -type GetAnalyticsCacheParams struct { - UserID string - Platform string - ToolName string - TimeRangeHash string -} - -func (q *Queries) GetAnalyticsCache(ctx context.Context, arg GetAnalyticsCacheParams) (json.RawMessage, error) { - row := q.db.QueryRowContext(ctx, getAnalyticsCache, - arg.UserID, - arg.Platform, - arg.ToolName, - arg.TimeRangeHash, - ) - var result json.RawMessage - err := row.Scan(&result) - return result, err -} - -const pruneAnalyticsCache = `-- name: PruneAnalyticsCache :exec -DELETE FROM analytics_cache WHERE expires_at <= now() -` - -func (q *Queries) PruneAnalyticsCache(ctx context.Context) error { - _, err := q.db.ExecContext(ctx, pruneAnalyticsCache) - return err -} - -const upsertAnalyticsCache = `-- name: UpsertAnalyticsCache :exec -INSERT INTO analytics_cache (user_id, platform, tool_name, time_range_hash, result, expires_at) -VALUES ($1, $2, $3, $4, $5, $6) -ON CONFLICT (user_id, platform, tool_name, time_range_hash) -DO UPDATE SET result = EXCLUDED.result, expires_at = EXCLUDED.expires_at, computed_at = now() -` - -type UpsertAnalyticsCacheParams struct { - UserID string - Platform string - ToolName string - TimeRangeHash string - Result json.RawMessage - ExpiresAt time.Time -} - -func (q *Queries) UpsertAnalyticsCache(ctx context.Context, arg UpsertAnalyticsCacheParams) error { - _, err := q.db.ExecContext(ctx, upsertAnalyticsCache, - arg.UserID, - arg.Platform, - arg.ToolName, - arg.TimeRangeHash, - arg.Result, - arg.ExpiresAt, - ) - return err -} diff --git a/internal/store/AnalyticsCoverage.sql.go b/internal/store/AnalyticsCoverage.sql.go deleted file mode 100644 index 3f8e0d1..0000000 --- a/internal/store/AnalyticsCoverage.sql.go +++ /dev/null @@ -1,59 +0,0 @@ -// Code generated by sqlc. DO NOT EDIT. -// versions: -// sqlc v1.31.1 -// source: AnalyticsCoverage.sql - -package store - -import ( - "context" - "time" -) - -const getAnalyticsCoverage = `-- name: GetAnalyticsCoverage :one -SELECT covered_start, covered_end FROM analytics_coverage -WHERE user_id = $1 AND platform = $2 -` - -type GetAnalyticsCoverageParams struct { - UserID string - Platform string -} - -type GetAnalyticsCoverageRow struct { - CoveredStart time.Time - CoveredEnd time.Time -} - -func (q *Queries) GetAnalyticsCoverage(ctx context.Context, arg GetAnalyticsCoverageParams) (GetAnalyticsCoverageRow, error) { - row := q.db.QueryRowContext(ctx, getAnalyticsCoverage, arg.UserID, arg.Platform) - var i GetAnalyticsCoverageRow - err := row.Scan(&i.CoveredStart, &i.CoveredEnd) - return i, err -} - -const upsertAnalyticsCoverage = `-- name: UpsertAnalyticsCoverage :exec -INSERT INTO analytics_coverage (user_id, platform, covered_start, covered_end, updated_at) -VALUES ($1, $2, $3, $4, now()) -ON CONFLICT (user_id, platform) -DO UPDATE SET covered_start = LEAST(analytics_coverage.covered_start, EXCLUDED.covered_start), - covered_end = GREATEST(analytics_coverage.covered_end, EXCLUDED.covered_end), - updated_at = now() -` - -type UpsertAnalyticsCoverageParams struct { - UserID string - Platform string - CoveredStart time.Time - CoveredEnd time.Time -} - -func (q *Queries) UpsertAnalyticsCoverage(ctx context.Context, arg UpsertAnalyticsCoverageParams) error { - _, err := q.db.ExecContext(ctx, upsertAnalyticsCoverage, - arg.UserID, - arg.Platform, - arg.CoveredStart, - arg.CoveredEnd, - ) - return err -} diff --git a/internal/store/AnalyticsPosts.sql.go b/internal/store/AnalyticsPosts.sql.go deleted file mode 100644 index 1cf59a1..0000000 --- a/internal/store/AnalyticsPosts.sql.go +++ /dev/null @@ -1,113 +0,0 @@ -// Code generated by sqlc. DO NOT EDIT. -// versions: -// sqlc v1.31.1 -// source: AnalyticsPosts.sql - -package store - -import ( - "context" - "time" -) - -const getAnalyticsPostsInRange = `-- name: GetAnalyticsPostsInRange :many -SELECT post_id, post_text, post_timestamp, likes, reposts, replies, quotes -FROM analytics_posts -WHERE user_id = $1 AND platform = $2 AND post_timestamp >= $3 AND post_timestamp <= $4 -ORDER BY post_timestamp DESC -LIMIT $5 -` - -type GetAnalyticsPostsInRangeParams struct { - UserID string - Platform string - PostTimestamp time.Time - PostTimestamp_2 time.Time - Limit int32 -} - -type GetAnalyticsPostsInRangeRow struct { - PostID string - PostText string - PostTimestamp time.Time - Likes int64 - Reposts int64 - Replies int64 - Quotes int64 -} - -func (q *Queries) GetAnalyticsPostsInRange(ctx context.Context, arg GetAnalyticsPostsInRangeParams) ([]GetAnalyticsPostsInRangeRow, error) { - rows, err := q.db.QueryContext(ctx, getAnalyticsPostsInRange, - arg.UserID, - arg.Platform, - arg.PostTimestamp, - arg.PostTimestamp_2, - arg.Limit, - ) - if err != nil { - return nil, err - } - defer rows.Close() - var items []GetAnalyticsPostsInRangeRow - for rows.Next() { - var i GetAnalyticsPostsInRangeRow - if err := rows.Scan( - &i.PostID, - &i.PostText, - &i.PostTimestamp, - &i.Likes, - &i.Reposts, - &i.Replies, - &i.Quotes, - ); err != nil { - return nil, err - } - items = append(items, i) - } - if err := rows.Close(); err != nil { - return nil, err - } - if err := rows.Err(); err != nil { - return nil, err - } - return items, nil -} - -const upsertAnalyticsPosts = `-- name: UpsertAnalyticsPosts :exec -INSERT INTO analytics_posts (user_id, platform, post_id, post_text, post_timestamp, likes, reposts, replies, quotes) -VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) -ON CONFLICT (user_id, platform, post_id) -DO UPDATE SET post_text = EXCLUDED.post_text, - post_timestamp = EXCLUDED.post_timestamp, - likes = EXCLUDED.likes, - reposts = EXCLUDED.reposts, - replies = EXCLUDED.replies, - quotes = EXCLUDED.quotes -` - -type UpsertAnalyticsPostsParams struct { - UserID string - Platform string - PostID string - PostText string - PostTimestamp time.Time - Likes int64 - Reposts int64 - Replies int64 - Quotes int64 -} - -func (q *Queries) UpsertAnalyticsPosts(ctx context.Context, arg UpsertAnalyticsPostsParams) error { - _, err := q.db.ExecContext(ctx, upsertAnalyticsPosts, - arg.UserID, - arg.Platform, - arg.PostID, - arg.PostText, - arg.PostTimestamp, - arg.Likes, - arg.Reposts, - arg.Replies, - arg.Quotes, - ) - return err -} diff --git a/internal/store/DeleteStylePosts.sql.go b/internal/store/DeleteStylePosts.sql.go deleted file mode 100644 index 6a8ce7b..0000000 --- a/internal/store/DeleteStylePosts.sql.go +++ /dev/null @@ -1,25 +0,0 @@ -// Code generated by sqlc. DO NOT EDIT. -// versions: -// sqlc v1.31.1 -// source: DeleteStylePosts.sql - -package store - -import ( - "context" -) - -const deleteStylePosts = `-- name: DeleteStylePosts :exec -DELETE FROM user_style_posts -WHERE user_id = $1 AND platform = $2 AND platform != 'Twitter' -` - -type DeleteStylePostsParams struct { - UserID string - Platform string -} - -func (q *Queries) DeleteStylePosts(ctx context.Context, arg DeleteStylePostsParams) error { - _, err := q.db.ExecContext(ctx, deleteStylePosts, arg.UserID, arg.Platform) - return err -} diff --git a/internal/store/FindSimilarUsers.sql.go b/internal/store/FindSimilarUsers.sql.go index ba1678e..638c23b 100644 --- a/internal/store/FindSimilarUsers.sql.go +++ b/internal/store/FindSimilarUsers.sql.go @@ -20,14 +20,14 @@ LIMIT $2 ` type FindSimilarUsersParams struct { - QueryEmbedding interface{} + QueryEmbedding any LimitVal int32 } type FindSimilarUsersRow struct { UserID string Platform string - Embedding interface{} + Embedding any TopKeywords []string StyleSummary string PostCount int32 diff --git a/internal/store/GetStylePostByID.sql.go b/internal/store/GetStylePostByID.sql.go deleted file mode 100644 index 2f6d77a..0000000 --- a/internal/store/GetStylePostByID.sql.go +++ /dev/null @@ -1,33 +0,0 @@ -// Code generated by sqlc. DO NOT EDIT. -// versions: -// sqlc v1.31.1 -// source: GetStylePostByID.sql - -package store - -import ( - "context" -) - -const getStylePostByID = `-- name: GetStylePostByID :one -SELECT post_id, post_text FROM user_style_posts -WHERE user_id = $1 AND platform = $2 AND post_id = $3 -` - -type GetStylePostByIDParams struct { - UserID string - Platform string - PostID string -} - -type GetStylePostByIDRow struct { - PostID string - PostText string -} - -func (q *Queries) GetStylePostByID(ctx context.Context, arg GetStylePostByIDParams) (GetStylePostByIDRow, error) { - row := q.db.QueryRowContext(ctx, getStylePostByID, arg.UserID, arg.Platform, arg.PostID) - var i GetStylePostByIDRow - err := row.Scan(&i.PostID, &i.PostText) - return i, err -} diff --git a/internal/store/GetStylePosts.sql.go b/internal/store/GetStylePosts.sql.go deleted file mode 100644 index e8872d4..0000000 --- a/internal/store/GetStylePosts.sql.go +++ /dev/null @@ -1,51 +0,0 @@ -// Code generated by sqlc. DO NOT EDIT. -// versions: -// sqlc v1.31.1 -// source: GetStylePosts.sql - -package store - -import ( - "context" -) - -const getStylePosts = `-- name: GetStylePosts :many -SELECT post_id, post_text FROM user_style_posts -WHERE user_id = $1 AND platform = $2 -ORDER BY created_at DESC -LIMIT $3 -` - -type GetStylePostsParams struct { - UserID string - Platform string - Limit int32 -} - -type GetStylePostsRow struct { - PostID string - PostText string -} - -func (q *Queries) GetStylePosts(ctx context.Context, arg GetStylePostsParams) ([]GetStylePostsRow, error) { - rows, err := q.db.QueryContext(ctx, getStylePosts, arg.UserID, arg.Platform, arg.Limit) - if err != nil { - return nil, err - } - defer rows.Close() - var items []GetStylePostsRow - for rows.Next() { - var i GetStylePostsRow - if err := rows.Scan(&i.PostID, &i.PostText); err != nil { - return nil, err - } - items = append(items, i) - } - if err := rows.Close(); err != nil { - return nil, err - } - if err := rows.Err(); err != nil { - return nil, err - } - return items, nil -} diff --git a/internal/store/GetStylePostsByIDs.sql.go b/internal/store/GetStylePostsByIDs.sql.go deleted file mode 100644 index b1996ab..0000000 --- a/internal/store/GetStylePostsByIDs.sql.go +++ /dev/null @@ -1,51 +0,0 @@ -// Code generated by sqlc. DO NOT EDIT. -// versions: -// sqlc v1.31.1 -// source: GetStylePostsByIDs.sql - -package store - -import ( - "context" - - "github.com/lib/pq" -) - -const getStylePostsByIDs = `-- name: GetStylePostsByIDs :many -SELECT post_id, post_text FROM user_style_posts -WHERE user_id = $1 AND platform = $2 AND post_id = ANY($3::text[]) -` - -type GetStylePostsByIDsParams struct { - UserID string - Platform string - Column3 []string -} - -type GetStylePostsByIDsRow struct { - PostID string - PostText string -} - -func (q *Queries) GetStylePostsByIDs(ctx context.Context, arg GetStylePostsByIDsParams) ([]GetStylePostsByIDsRow, error) { - rows, err := q.db.QueryContext(ctx, getStylePostsByIDs, arg.UserID, arg.Platform, pq.Array(arg.Column3)) - if err != nil { - return nil, err - } - defer rows.Close() - var items []GetStylePostsByIDsRow - for rows.Next() { - var i GetStylePostsByIDsRow - if err := rows.Scan(&i.PostID, &i.PostText); err != nil { - return nil, err - } - items = append(items, i) - } - if err := rows.Close(); err != nil { - return nil, err - } - if err := rows.Err(); err != nil { - return nil, err - } - return items, nil -} diff --git a/internal/store/PruneStylePosts.sql.go b/internal/store/PruneStylePosts.sql.go deleted file mode 100644 index 52a36c8..0000000 --- a/internal/store/PruneStylePosts.sql.go +++ /dev/null @@ -1,20 +0,0 @@ -// Code generated by sqlc. DO NOT EDIT. -// versions: -// sqlc v1.31.1 -// source: PruneStylePosts.sql - -package store - -import ( - "context" -) - -const pruneStylePosts = `-- name: PruneStylePosts :exec -DELETE FROM user_style_posts -WHERE created_at < now() - interval '7 days' AND platform != 'Twitter' -` - -func (q *Queries) PruneStylePosts(ctx context.Context) error { - _, err := q.db.ExecContext(ctx, pruneStylePosts) - return err -} diff --git a/internal/store/UpsertStylePosts.sql.go b/internal/store/UpsertStylePosts.sql.go deleted file mode 100644 index 8839ffe..0000000 --- a/internal/store/UpsertStylePosts.sql.go +++ /dev/null @@ -1,34 +0,0 @@ -// Code generated by sqlc. DO NOT EDIT. -// versions: -// sqlc v1.31.1 -// source: UpsertStylePosts.sql - -package store - -import ( - "context" -) - -const upsertStylePosts = `-- name: UpsertStylePosts :exec -INSERT INTO user_style_posts (user_id, platform, post_id, post_text) -VALUES ($1, $2, $3, $4) -ON CONFLICT (user_id, platform, post_id) -DO UPDATE SET post_text = EXCLUDED.post_text, created_at = now() -` - -type UpsertStylePostsParams struct { - UserID string - Platform string - PostID string - PostText string -} - -func (q *Queries) UpsertStylePosts(ctx context.Context, arg UpsertStylePostsParams) error { - _, err := q.db.ExecContext(ctx, upsertStylePosts, - arg.UserID, - arg.Platform, - arg.PostID, - arg.PostText, - ) - return err -} diff --git a/internal/store/UpsertStyleProfiles.sql.go b/internal/store/UpsertStyleProfiles.sql.go index 7dadce2..1a53dd3 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 interface{} + Embedding any TopKeywords []string StyleSummary string PostCount int32 diff --git a/internal/store/db.go b/internal/store/db.go index fd993d9..6bf74de 100644 --- a/internal/store/db.go +++ b/internal/store/db.go @@ -10,10 +10,10 @@ import ( ) type DBTX interface { - ExecContext(context.Context, string, ...interface{}) (sql.Result, error) + ExecContext(context.Context, string, ...any) (sql.Result, error) PrepareContext(context.Context, string) (*sql.Stmt, error) - QueryContext(context.Context, string, ...interface{}) (*sql.Rows, error) - QueryRowContext(context.Context, string, ...interface{}) *sql.Row + QueryContext(context.Context, string, ...any) (*sql.Rows, error) + QueryRowContext(context.Context, string, ...any) *sql.Row } func New(db DBTX) *Queries { diff --git a/internal/store/integration_test.go b/internal/store/integration_test.go index 820fafe..7f1cedf 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,", 1024), ",") + "]" + emb := "[" + strings.TrimSuffix(strings.Repeat("0.0,", 768), ",") + "]" if err := q.UpsertStyleProfile(ctx, UpsertStyleProfileParams{ UserID: uid, diff --git a/internal/store/models.go b/internal/store/models.go index d3820d8..c62215b 100644 --- a/internal/store/models.go +++ b/internal/store/models.go @@ -1,48 +1,13 @@ // Code generated by sqlc. DO NOT EDIT. // versions: -// sqlc v1.31.1 +// sqlc v1.30.0 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 @@ -54,20 +19,11 @@ 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 interface{} + Embedding any TopKeywords []string StyleSummary string PostCount int32 diff --git a/internal/utils/styleProfiles.go b/internal/utils/styleProfiles.go index 7a6a587..27c372d 100644 --- a/internal/utils/styleProfiles.go +++ b/internal/utils/styleProfiles.go @@ -38,9 +38,11 @@ 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: @@ -74,7 +76,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.FieldsSeq(strings.ToLower(p)) { + for _, w := range strings.Fields(strings.ToLower(p)) { if len(w) < 4 { continue } diff --git a/scripts/fund-wallet.sh b/scripts/fund-wallet.sh deleted file mode 100755 index 7ad808d..0000000 --- a/scripts/fund-wallet.sh +++ /dev/null @@ -1,62 +0,0 @@ -#!/bin/bash -set -euo pipefail - -echo "=== Funding Wallet with Base Sepolia USDC ===" - -# Required environment variables -: "${PRIVATE_KEY:?PRIVATE_KEY must be set}" -: "${FUNDING_FAUCET_URL:?FUNDING_FAUCET_URL must be set}" - -WALLET_ADDRESS=$(cast wallet address --private-key "$PRIVATE_KEY") -echo "Wallet address: $WALLET_ADDRESS" - -# Check current USDC balance -USDC_CONTRACT="0x036CbD53842c5426634e7929541eC2318f3dCF7e" -RPC_URL="https://sepolia.base.org" - -BALANCE=$(cast call "$USDC_CONTRACT" "balanceOf(address)(uint256)" "$WALLET_ADDRESS" --rpc-url "$RPC_URL" 2>/dev/null || echo "0") -BALANCE_DECIMAL=$(cast to-dec "$BALANCE" 2>/dev/null || echo "0") -BALANCE_USDC=$(echo "scale=6; $BALANCE_DECIMAL / 1000000" | bc 2>/dev/null || echo "0") - -echo "Current USDC balance: $BALANCE_USDC USDC" - -# If balance is less than 10 USDC, request from faucet -if (( $(echo "$BALANCE_USDC < 10" | bc -l) )); then - echo "Balance below 10 USDC, requesting from faucet..." - - # Request from faucet (Coinbase CDP faucet API format) - curl -X POST "$FUNDING_FAUCET_URL" \ - -H "Content-Type: application/json" \ - -d "{\"address\": \"$WALLET_ADDRESS\", \"token\": \"USDC\"}" \ - || echo "Faucet request failed, may need manual funding" - - # Wait for funding - echo "Waiting for funding..." - sleep 30 - - # Check balance again - BALANCE=$(cast call "$USDC_CONTRACT" "balanceOf(address)(uint256)" "$WALLET_ADDRESS" --rpc-url "$RPC_URL" 2>/dev/null || echo "0") - BALANCE_DECIMAL=$(cast to-dec "$BALANCE" 2>/dev/null || echo "0") - BALANCE_USDC=$(echo "scale=6; $BALANCE_DECIMAL / 1000000" | bc 2>/dev/null || echo "0") - echo "New USDC balance: $BALANCE_USDC USDC" -else - echo "Sufficient USDC balance" -fi - -# Also fund with ETH for gas -ETH_BALANCE=$(cast balance "$WALLET_ADDRESS" --rpc-url "$RPC_URL" 2>/dev/null || echo "0") -ETH_BALANCE_DECIMAL=$(cast to-dec "$ETH_BALANCE" 2>/dev/null || echo "0") -ETH_BALANCE_ETH=$(echo "scale=6; $ETH_BALANCE_DECIMAL / 1000000000000000000" | bc 2>/dev/null || echo "0") - -echo "Current ETH balance: $ETH_BALANCE_ETH ETH" - -if (( $(echo "$ETH_BALANCE_ETH < 0.01" | bc -l) )); then - echo "ETH balance low, requesting from faucet..." - curl -X POST "$FUNDING_FAUCET_URL" \ - -H "Content-Type: application/json" \ - -d "{\"address\": \"$WALLET_ADDRESS\", \"token\": \"ETH\"}" \ - || echo "ETH faucet request failed" - sleep 10 -fi - -echo "=== Wallet Funding Complete ===" \ No newline at end of file diff --git a/scripts/start-test-server.sh b/scripts/start-test-server.sh deleted file mode 100755 index cdee795..0000000 --- a/scripts/start-test-server.sh +++ /dev/null @@ -1,83 +0,0 @@ -#!/bin/bash -set -euo pipefail - -echo "=== Starting Test Server ===" - -# Required environment variables -: "${PRIVATE_KEY:?PRIVATE_KEY must be set}" -: "${X402_WALLET_ADDRESS:?X402_WALLET_ADDRESS must be set}" - -# Set up environment for local testing -export SERVER_ADDRESS="127.0.0.1" -export SERVER_PORT="8080" -export GOOSE_DBSTRING="postgres://postgres:postgres@localhost:5432/mcpserver?sslmode=disable" -export ACTIVE_NETWORKS="sepolia" -export X402_FACILITATOR_URL="https://api.cdp.coinbase.com/platform/v2/x402" -export X402_WALLET_ADDRESS="$X402_WALLET_ADDRESS" -export OLLAMA_HOST="http://localhost:11434" -export MIGRATE_ON_START="true" -export MIGRATIONS_DIR="sql/schema" -export READYZ_PORT="9090" -export METRICS_PORT="9091" -export LOG_FILE="/tmp/mcpserver-test.log" -export LOG_LEVEL="debug" - -# Use test credentials or empty for local testing -export BLUESKY_DID="${BLUESKY_DID:-}" -export BLUESKY_APP_PASSWORD="${BLUESKY_APP_PASSWORD:-}" -export BLUESKY_HOST="${BLUESKY_HOST:-https://bsky.social}" -export AP_HOST="${AP_HOST:-https://mastodon.social}" -export AP_ACCESS_TOKEN="${AP_ACCESS_TOKEN:-}" -export NOSTR_RELAYS="${NOSTR_RELAYS:-wss://relay.damus.io,wss://nos.lol}" -export TWITTER_BEARER_TOKEN="${TWITTER_BEARER_TOKEN:-}" -export CDP_API_KEY_ID="${CDP_API_KEY_ID:-}" -export CDP_API_SECRET="${CDP_API_SECRET:-}" -export ENRICH_CROSS_PLATFORM="false" -export PREWARM_AUTHORS="" -export PREWARM_FEED_URI="" - -# Start the server in background -echo "Starting server on port 8080..." -go run ./cmd/server/main.go & -SERVER_PID=$! - -# Save PID for cleanup -echo $SERVER_PID > /tmp/test-server.pid - -# Wait for server to be ready -echo "Waiting for server to be ready..." -for i in {1..30}; do - if curl -sf http://localhost:9090/readyz >/dev/null 2>&1; then - echo "Server is ready" - break - fi - sleep 1 -done - -# Verify server is responding -curl -sf http://localhost:8080/healthz >/dev/null 2>&1 || { - echo "Server health check failed" - kill $SERVER_PID 2>/dev/null || true - exit 1 -} - -# Verify payment endpoints are working (should return 402 without payment) -echo "Verifying payment endpoints..." -if curl -sf -X POST http://localhost:8080/v1/getActor \ - -H "Content-Type: application/json" \ - -d '{"author": "did:plc:z72i7hdynmk6r22z27h6tvur"}' >/dev/null 2>&1; then - echo "WARNING: Payable endpoint getActor returned 200 without payment!" -else - echo "Payment endpoint getActor correctly requires payment (402)" -fi - -# Verify free endpoint works -if curl -sf -X POST http://localhost:8080/v1/getFeed \ - -H "Content-Type: application/json" \ - -d '{"uri": "https://bsky.app/profile/did:plc:z72i7hdynmk6r22z27h6tvur/feed/whats-hot"}' >/dev/null 2>&1; then - echo "Free endpoint getFeed works correctly" -else - echo "WARNING: Free endpoint getFeed failed!" -fi - -echo "=== Test Server Started (PID: $SERVER_PID) ===" \ No newline at end of file diff --git a/sql/queries/AnalyticsCache.sql b/sql/queries/AnalyticsCache.sql deleted file mode 100644 index 670da7e..0000000 --- a/sql/queries/AnalyticsCache.sql +++ /dev/null @@ -1,17 +0,0 @@ --- name: GetAnalyticsCache :one -SELECT result FROM analytics_cache -WHERE user_id = $1 AND platform = $2 AND tool_name = $3 AND time_range_hash = $4 - AND expires_at > now(); - --- name: UpsertAnalyticsCache :exec -INSERT INTO analytics_cache (user_id, platform, tool_name, time_range_hash, result, expires_at) -VALUES ($1, $2, $3, $4, $5, $6) -ON CONFLICT (user_id, platform, tool_name, time_range_hash) -DO UPDATE SET result = EXCLUDED.result, expires_at = EXCLUDED.expires_at, computed_at = now(); - --- name: DeleteAnalyticsCache :exec -DELETE FROM analytics_cache -WHERE user_id = $1 AND platform = $2 AND tool_name = $3 AND time_range_hash = $4; - --- name: PruneAnalyticsCache :exec -DELETE FROM analytics_cache WHERE expires_at <= now(); \ No newline at end of file diff --git a/sql/queries/AnalyticsCoverage.sql b/sql/queries/AnalyticsCoverage.sql deleted file mode 100644 index 58ad395..0000000 --- a/sql/queries/AnalyticsCoverage.sql +++ /dev/null @@ -1,11 +0,0 @@ --- name: GetAnalyticsCoverage :one -SELECT covered_start, covered_end FROM analytics_coverage -WHERE user_id = $1 AND platform = $2; - --- name: UpsertAnalyticsCoverage :exec -INSERT INTO analytics_coverage (user_id, platform, covered_start, covered_end, updated_at) -VALUES ($1, $2, $3, $4, now()) -ON CONFLICT (user_id, platform) -DO UPDATE SET covered_start = LEAST(analytics_coverage.covered_start, EXCLUDED.covered_start), - covered_end = GREATEST(analytics_coverage.covered_end, EXCLUDED.covered_end), - updated_at = now(); diff --git a/sql/queries/AnalyticsPosts.sql b/sql/queries/AnalyticsPosts.sql deleted file mode 100644 index 6f7c31e..0000000 --- a/sql/queries/AnalyticsPosts.sql +++ /dev/null @@ -1,17 +0,0 @@ --- name: GetAnalyticsPostsInRange :many -SELECT post_id, post_text, post_timestamp, likes, reposts, replies, quotes -FROM analytics_posts -WHERE user_id = $1 AND platform = $2 AND post_timestamp >= $3 AND post_timestamp <= $4 -ORDER BY post_timestamp DESC -LIMIT $5; - --- name: UpsertAnalyticsPosts :exec -INSERT INTO analytics_posts (user_id, platform, post_id, post_text, post_timestamp, likes, reposts, replies, quotes) -VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) -ON CONFLICT (user_id, platform, post_id) -DO UPDATE SET post_text = EXCLUDED.post_text, - post_timestamp = EXCLUDED.post_timestamp, - likes = EXCLUDED.likes, - reposts = EXCLUDED.reposts, - replies = EXCLUDED.replies, - quotes = EXCLUDED.quotes; diff --git a/sql/queries/DeleteStylePosts.sql b/sql/queries/DeleteStylePosts.sql deleted file mode 100644 index 163eaae..0000000 --- a/sql/queries/DeleteStylePosts.sql +++ /dev/null @@ -1,3 +0,0 @@ --- 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 deleted file mode 100644 index 255531c..0000000 --- a/sql/queries/GetStylePostByID.sql +++ /dev/null @@ -1,3 +0,0 @@ --- name: GetStylePostByID :one -SELECT post_id, post_text FROM user_style_posts -WHERE user_id = $1 AND platform = $2 AND post_id = $3; \ No newline at end of file diff --git a/sql/queries/GetStylePosts.sql b/sql/queries/GetStylePosts.sql deleted file mode 100644 index 3c1ca49..0000000 --- a/sql/queries/GetStylePosts.sql +++ /dev/null @@ -1,5 +0,0 @@ --- name: GetStylePosts :many -SELECT post_id, post_text FROM user_style_posts -WHERE user_id = $1 AND platform = $2 -ORDER BY created_at DESC -LIMIT $3; \ No newline at end of file diff --git a/sql/queries/GetStylePostsByIDs.sql b/sql/queries/GetStylePostsByIDs.sql deleted file mode 100644 index 932b794..0000000 --- a/sql/queries/GetStylePostsByIDs.sql +++ /dev/null @@ -1,3 +0,0 @@ --- name: GetStylePostsByIDs :many -SELECT post_id, post_text FROM user_style_posts -WHERE user_id = $1 AND platform = $2 AND post_id = ANY($3::text[]); \ No newline at end of file diff --git a/sql/queries/PruneStylePosts.sql b/sql/queries/PruneStylePosts.sql deleted file mode 100644 index 064140d..0000000 --- a/sql/queries/PruneStylePosts.sql +++ /dev/null @@ -1,3 +0,0 @@ --- 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 deleted file mode 100644 index 2c669bf..0000000 --- a/sql/queries/UpsertStylePosts.sql +++ /dev/null @@ -1,5 +0,0 @@ --- name: UpsertStylePosts :exec -INSERT INTO user_style_posts (user_id, platform, post_id, post_text) -VALUES ($1, $2, $3, $4) -ON CONFLICT (user_id, platform, post_id) -DO UPDATE SET post_text = EXCLUDED.post_text, created_at = now(); \ No newline at end of file diff --git a/sql/schema/001_transactions.sql b/sql/schema/001_transactions.sql index 612215f..700b6b8 100644 --- a/sql/schema/001_transactions.sql +++ b/sql/schema/001_transactions.sql @@ -1,5 +1,5 @@ -- +goose up -CREATE TABLE IF NOT EXISTS transactions ( +CREATE TABLE 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 6241a7e..14d6307 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 IF NOT EXISTS user_style_profiles ( +CREATE TABLE user_style_profiles ( id BIGSERIAL PRIMARY KEY, user_id TEXT NOT NULL, platform TEXT NOT NULL, - embedding VECTOR(1024) NOT NULL, + embedding VECTOR(768) 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 4c86c15..42172d6 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 IF NOT EXISTS idx_user_style_profiles_versions +CREATE INDEX idx_user_style_profiles_versions ON user_style_profiles (model_version, schema_version, algorithm_version); -- +goose Down diff --git a/sql/schema/004_add_style_posts.sql b/sql/schema/004_add_style_posts.sql deleted file mode 100644 index cc020c1..0000000 --- a/sql/schema/004_add_style_posts.sql +++ /dev/null @@ -1,17 +0,0 @@ --- +goose up -CREATE TABLE IF NOT EXISTS user_style_posts ( - id BIGSERIAL PRIMARY KEY, - user_id TEXT NOT NULL, - platform TEXT NOT NULL, - post_id TEXT NOT NULL, - post_text TEXT NOT NULL, - created_at TIMESTAMPTZ NOT NULL DEFAULT now(), - UNIQUE (user_id, platform, post_id) -); - -CREATE INDEX idx_user_style_posts_user_platform - ON user_style_posts (user_id, platform); - --- +goose down -DROP INDEX IF EXISTS idx_user_style_posts_user_platform; -DROP TABLE IF EXISTS user_style_posts; \ No newline at end of file diff --git a/sql/schema/005_add_analytics_cache.sql b/sql/schema/005_add_analytics_cache.sql deleted file mode 100644 index 3dd106c..0000000 --- a/sql/schema/005_add_analytics_cache.sql +++ /dev/null @@ -1,23 +0,0 @@ --- +goose up -CREATE TABLE IF NOT EXISTS analytics_cache ( - id BIGSERIAL PRIMARY KEY, - user_id TEXT NOT NULL, - platform TEXT NOT NULL, - tool_name TEXT NOT NULL, - time_range_hash TEXT NOT NULL, - result JSONB NOT NULL, - computed_at TIMESTAMPTZ NOT NULL DEFAULT now(), - expires_at TIMESTAMPTZ NOT NULL, - UNIQUE (user_id, platform, tool_name, time_range_hash) -); - -CREATE INDEX IF NOT EXISTS idx_analytics_cache_lookup - ON analytics_cache (user_id, platform, tool_name, time_range_hash); - -CREATE INDEX IF NOT EXISTS idx_analytics_cache_expires - ON analytics_cache (expires_at); - --- +goose down -DROP INDEX IF EXISTS idx_analytics_cache_expires; -DROP INDEX IF EXISTS idx_analytics_cache_lookup; -DROP TABLE IF EXISTS analytics_cache; \ No newline at end of file diff --git a/sql/schema/006_add_analytics_posts.sql b/sql/schema/006_add_analytics_posts.sql deleted file mode 100644 index 36fc965..0000000 --- a/sql/schema/006_add_analytics_posts.sql +++ /dev/null @@ -1,22 +0,0 @@ --- +goose up -CREATE TABLE IF NOT EXISTS analytics_posts ( - id BIGSERIAL PRIMARY KEY, - user_id TEXT NOT NULL, - platform TEXT NOT NULL, - post_id TEXT NOT NULL, - post_text TEXT NOT NULL, - post_timestamp TIMESTAMPTZ NOT NULL, - likes BIGINT NOT NULL DEFAULT 0, - reposts BIGINT NOT NULL DEFAULT 0, - replies BIGINT NOT NULL DEFAULT 0, - quotes BIGINT NOT NULL DEFAULT 0, - created_at TIMESTAMPTZ NOT NULL DEFAULT now(), - UNIQUE (user_id, platform, post_id) -); - -CREATE INDEX IF NOT EXISTS idx_analytics_posts_user_platform_time - ON analytics_posts (user_id, platform, post_timestamp); - --- +goose down -DROP INDEX IF EXISTS idx_analytics_posts_user_platform_time; -DROP TABLE IF EXISTS analytics_posts; diff --git a/sql/schema/007_add_analytics_coverage.sql b/sql/schema/007_add_analytics_coverage.sql deleted file mode 100644 index b22ac24..0000000 --- a/sql/schema/007_add_analytics_coverage.sql +++ /dev/null @@ -1,12 +0,0 @@ --- +goose up -CREATE TABLE IF NOT EXISTS analytics_coverage ( - user_id TEXT NOT NULL, - platform TEXT NOT NULL, - covered_start TIMESTAMPTZ NOT NULL, - covered_end TIMESTAMPTZ NOT NULL, - updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), - PRIMARY KEY (user_id, platform) -); - --- +goose down -DROP TABLE IF EXISTS analytics_coverage; diff --git a/testClient/main.go b/testClient/main.go index 2e77c1f..394d2c1 100644 --- a/testClient/main.go +++ b/testClient/main.go @@ -1,22 +1,16 @@ package main import ( + "bytes" "context" - "encoding/json" "fmt" "io" "log" - "math/big" + "os" + "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" @@ -25,15 +19,6 @@ 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 } @@ -46,6 +31,7 @@ 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) @@ -53,11 +39,13 @@ 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) @@ -68,206 +56,13 @@ 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() - 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) - } + var ( + privateKey = os.Getenv("PRIVATE_KEY") + serverURL = "https://api.naco.li/v1" + ctx = context.Background() + ) signer, err := evmsigners.NewClientSignerFromPrivateKey(privateKey) if err != nil { @@ -275,13 +70,12 @@ func main() { } x402Client := x402.Newx402Client(). - Register("eip155:84532", evm.NewExactEvmScheme(signer, nil)) + Register("eip155:*", evm.NewExactEvmScheme(signer, nil)) debugClient := &http.Client{ Transport: &LoggingRoundTripper{ Proxied: http.DefaultTransport, }, - Timeout: 60 * time.Second, } httpClient := x402http.WrapHTTPClientWithPayment( @@ -289,44 +83,39 @@ func main() { x402http.Newx402HTTPClient(x402Client), ) - log.Printf("Starting REST API endpoint tests against %s...", serverURL) + log.Println("Starting REST API tests...") - 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 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)) } - passed := 0 - failed := 0 + log.Println("\nCalling getPost (gated paid tool) via HTTP client...") + bodyGetPost := []byte(`{"actor":"sharkgamiart.bsky.social","rkey":"3m47y27c67c2b"}`) - 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, err := http.NewRequestWithContext(ctx, "POST", fmt.Sprintf("%s/getPost", serverURL), bytes.NewReader(bodyGetPost)) + if err != nil { + log.Fatal("Failed to create getPost request:", err) } + req.GetBody = func() (io.ReadCloser, error) { + return io.NopCloser(bytes.NewReader(bodyGetPost)), nil + } + req.Header.Set("Content-Type", "application/json") - log.Printf("\n=== Test Summary ===") - log.Printf("Passed: %d", passed) - log.Printf("Failed: %d", failed) - - if failed > 0 { - os.Exit(1) + 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)) } }