From 9d5e7a94f6be03977e801ecb6760a93012327e95 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 15 Jul 2026 07:24:00 +0000 Subject: [PATCH 1/2] [SO-279] Engagement Tracking, Airdrop Points & Leaderboard MVP MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Engagement→airdrop pipeline (design doc: docs/engagement-airdrop-plan.md): - twitter-service: signed read endpoints GET /mentions (recent search for tweets mentioning an account) and GET /tweets/metrics (refresh counters for up to 100 tweets), sharing the existing OAuth 1.0a signer. - engagement-service (new, internal :9017): polls twitter-service for mentions, persists per-tweet like/retweet/reply/quote counters in Postgres (diesel + embedded migrations), derives engagement/airdrop points from config weights (ambassador multiplier), serves GET /leaderboard and GET /points/{handle}. - airdrop-bot (new, public :9018): Discord slash-command bot — a separate Discord application + deployment from social-bot — with read-only /leaderboard and /points commands served from engagement-service. - Deployment wiring (staging-only, like twitter-service/social-bot): Dockerfiles, bake.hcl targets, affected.py, deploy.sh, render-secrets.sh (options//airdrop-bot), compose entries, nginx route, terraform secret placeholder. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01CacZTy2eX51YUrMNcnpKm7 --- .github/workflows/_deploy.yml | 2 +- docs/engagement-airdrop-plan.md | 97 ++++++++ rust-backend/Cargo.lock | 41 ++++ rust-backend/Cargo.toml | 2 + rust-backend/Dockerfile.airdrop-bot | 30 +++ rust-backend/Dockerfile.engagement-service | 30 +++ rust-backend/deployment/affected.py | 18 +- rust-backend/deployment/bake.hcl | 18 +- .../compose/docker-compose.staging.yml | 36 ++- rust-backend/deployment/ec2/deploy.sh | 7 +- rust-backend/deployment/ec2/render-secrets.sh | 20 ++ .../deployment/nginx/nginx.staging.conf | 12 + rust-backend/infra/secrets.tf | 25 +++ rust-backend/services/airdrop-bot/Cargo.toml | 39 ++++ rust-backend/services/airdrop-bot/README.md | 65 ++++++ .../airdrop-bot/config/config.staging.toml | 8 + .../services/airdrop-bot/config/config.toml | 12 + .../airdrop-bot/config/secrets.example.toml | 8 + .../services/airdrop-bot/src/commands.rs | 127 +++++++++++ .../services/airdrop-bot/src/config.rs | 25 +++ .../services/airdrop-bot/src/discord.rs | 195 ++++++++++++++++ .../airdrop-bot/src/engagement_client.rs | 90 ++++++++ rust-backend/services/airdrop-bot/src/lib.rs | 62 ++++++ rust-backend/services/airdrop-bot/src/main.rs | 45 ++++ .../services/airdrop-bot/src/router.rs | 33 +++ .../services/airdrop-bot/src/secrets.rs | 28 +++ .../services/airdrop-bot/src/state.rs | 14 ++ .../services/engagement-service/Cargo.toml | 40 ++++ .../services/engagement-service/README.md | 67 ++++++ .../config/config.staging.toml | 32 +++ .../engagement-service/config/config.toml | 31 +++ .../services/engagement-service/src/config.rs | 60 +++++ .../src/db/migrations/000001_init/down.sql | 2 + .../src/db/migrations/000001_init/up.sql | 26 +++ .../services/engagement-service/src/db/mod.rs | 30 +++ .../engagement-service/src/db/models.rs | 30 +++ .../engagement-service/src/db/repo.rs | 165 ++++++++++++++ .../engagement-service/src/db/schema.rs | 32 +++ .../engagement-service/src/handlers.rs | 168 ++++++++++++++ .../services/engagement-service/src/lib.rs | 59 +++++ .../services/engagement-service/src/main.rs | 41 ++++ .../services/engagement-service/src/points.rs | 122 ++++++++++ .../services/engagement-service/src/poller.rs | 109 +++++++++ .../services/engagement-service/src/router.rs | 29 +++ .../services/engagement-service/src/state.rs | 11 + .../engagement-service/src/twitter_client.rs | 99 +++++++++ .../services/twitter-service/src/handlers.rs | 165 +++++++++++++- .../services/twitter-service/src/lib.rs | 4 + .../services/twitter-service/src/router.rs | 2 + .../services/twitter-service/src/twitter.rs | 209 +++++++++++++++++- 50 files changed, 2606 insertions(+), 16 deletions(-) create mode 100644 docs/engagement-airdrop-plan.md create mode 100644 rust-backend/Dockerfile.airdrop-bot create mode 100644 rust-backend/Dockerfile.engagement-service create mode 100644 rust-backend/services/airdrop-bot/Cargo.toml create mode 100644 rust-backend/services/airdrop-bot/README.md create mode 100644 rust-backend/services/airdrop-bot/config/config.staging.toml create mode 100644 rust-backend/services/airdrop-bot/config/config.toml create mode 100644 rust-backend/services/airdrop-bot/config/secrets.example.toml create mode 100644 rust-backend/services/airdrop-bot/src/commands.rs create mode 100644 rust-backend/services/airdrop-bot/src/config.rs create mode 100644 rust-backend/services/airdrop-bot/src/discord.rs create mode 100644 rust-backend/services/airdrop-bot/src/engagement_client.rs create mode 100644 rust-backend/services/airdrop-bot/src/lib.rs create mode 100644 rust-backend/services/airdrop-bot/src/main.rs create mode 100644 rust-backend/services/airdrop-bot/src/router.rs create mode 100644 rust-backend/services/airdrop-bot/src/secrets.rs create mode 100644 rust-backend/services/airdrop-bot/src/state.rs create mode 100644 rust-backend/services/engagement-service/Cargo.toml create mode 100644 rust-backend/services/engagement-service/README.md create mode 100644 rust-backend/services/engagement-service/config/config.staging.toml create mode 100644 rust-backend/services/engagement-service/config/config.toml create mode 100644 rust-backend/services/engagement-service/src/config.rs create mode 100644 rust-backend/services/engagement-service/src/db/migrations/000001_init/down.sql create mode 100644 rust-backend/services/engagement-service/src/db/migrations/000001_init/up.sql create mode 100644 rust-backend/services/engagement-service/src/db/mod.rs create mode 100644 rust-backend/services/engagement-service/src/db/models.rs create mode 100644 rust-backend/services/engagement-service/src/db/repo.rs create mode 100644 rust-backend/services/engagement-service/src/db/schema.rs create mode 100644 rust-backend/services/engagement-service/src/handlers.rs create mode 100644 rust-backend/services/engagement-service/src/lib.rs create mode 100644 rust-backend/services/engagement-service/src/main.rs create mode 100644 rust-backend/services/engagement-service/src/points.rs create mode 100644 rust-backend/services/engagement-service/src/poller.rs create mode 100644 rust-backend/services/engagement-service/src/router.rs create mode 100644 rust-backend/services/engagement-service/src/state.rs create mode 100644 rust-backend/services/engagement-service/src/twitter_client.rs diff --git a/.github/workflows/_deploy.yml b/.github/workflows/_deploy.yml index 853923ac..fc439f1a 100644 --- a/.github/workflows/_deploy.yml +++ b/.github/workflows/_deploy.yml @@ -156,7 +156,7 @@ jobs: # Keep this in sync with ALL_SERVICES in deployment/ec2/deploy.sh # and affected.py — a service missing here is never tag-seeded on # force_all, so deploy.sh rejects the deploy on a fresh box. - services='["indexer","quoting-service","mm-bot","option-scheduler","api-service","token-info","auth-service","gas-station","price-charting","balance-monitor","keeper","oracle-service","cctp-relay","twitter-service","social-bot"]' + services='["indexer","quoting-service","mm-bot","option-scheduler","api-service","token-info","auth-service","gas-station","price-charting","balance-monitor","keeper","oracle-service","cctp-relay","twitter-service","social-bot","engagement-service","airdrop-bot"]' echo "force_all or image_tag override → rolling all services" else services=$(printf '%s\n' "$CHANGED_FILES" \ diff --git a/docs/engagement-airdrop-plan.md b/docs/engagement-airdrop-plan.md new file mode 100644 index 00000000..caff2066 --- /dev/null +++ b/docs/engagement-airdrop-plan.md @@ -0,0 +1,97 @@ +# Engagement → Airdrop Leaderboard + +Plan + MVP scope for the engagement/airdrop program: track engagement on +tweets that mention us, convert it into airdrop points, rank ambassadors and +the general public on a leaderboard, and expose it all through a Discord bot. + +## Architecture + +``` + Twitter API v2 (OAuth 1.0a, per-account) + ▲ + │ POST /2/tweets (existing) + │ GET /2/tweets/search/recent (new) + │ GET /2/tweets?ids=… (new) + ┌───────┴───────┐ + │ twitter-service│ internal :9014 + └───┬───────▲───┘ + GET /mentions │ │ (also: social-bot /tweet, + GET /tweets/metrics │ unchanged) + ┌───▼───────┴───┐ + │ engagement- │ internal :9017 + │ service │──▶ Postgres engagement_ + │ poll loop │ (tracked_tweets, + │ points module │ poll_cursor) + └───┬───────────┘ + GET /leaderboard │ GET /points/{handle} + ┌───▼───────────┐ + │ airdrop-bot │ public :9018 + │ (Discord app, │ nginx //airdrop-bot/ + │ separate from │ + │ social-bot) │ + └───────────────┘ +``` + +Four pieces from the original sketch, mapped to what shipped: + +1. **Engagement tracking** — twitter-service grew two signed read endpoints + (`GET /mentions`, `GET /tweets/metrics`); it stays the single owner of + Twitter credentials. engagement-service polls it every 5 minutes: new + mentions of `@suioptions` (original tweets only, no retweets, last 7 + days — Twitter's recent-search window) are upserted into Postgres, and + the stalest counters among tweets younger than 7 days are refreshed 100 + at a time. +2. **Airdrop conversion** — a pure module inside engagement-service, not a + separate service (see "Decisions"). Config weights per like/reply/ + retweet/quote → engagement points; a rate + ambassador multiplier → + airdrop points. Derived at read time, so tuning weights or the + ambassador roster is a config deploy, no backfill. +3. **Leaderboard** — `GET /leaderboard` on engagement-service: authors + (keyed by twitter handle) ranked by airdrop points; ambassadors are + flagged, the general public competes on the same board. +4. **Discord bot** — airdrop-bot, a separate Discord application and + container from social-bot (the team tweeting bot): community-facing, + read-only `/leaderboard [count]` and `/points ` commands, no + allow list, no shared secrets. + +## Decisions (and why) + +- **One service for tracking + conversion + leaderboard.** The partner + sketch has engagement-tracking and airdrop-tracking as two services, but + the conversion is a pure function over the tracked totals — a second + service would add a network hop, a deployment and a DB with no state of + its own. `points.rs` keeps the boundary; split it out when the airdrop + needs real state (claims, epochs, on-chain distribution). +- **Points are derived, never stored.** Only raw counters live in the DB. +- **Handles, not user ids, key the leaderboard.** Simpler to read, matches + the ambassador roster in config; `author_id` is stored per tweet so we + can re-key later if handle changes ever matter. +- **airdrop-bot is webhook-based like social-bot** (no gateway connection): + one axum server, Ed25519-verified interactions, defer + follow-up within + Discord's 3s ack window. +- **Staging-only for now**, like twitter-service/social-bot — deliberately + not declared in docker-compose.prod.yml. + +## MVP scoping (not built, by design) + +- Engagement history/snapshots (trend charts), follower-weighting, + spam/bot filtering beyond excluding retweets. +- Mentions older than Twitter's 7-day recent-search window: the poll cursor + makes this moot once the service is running continuously. +- Claim flow / on-chain distribution — the leaderboard is the deliverable; + distribution is a later phase with its own design. +- Discord↔Twitter account linking (`/register`): points accrue to twitter + handles; anyone can query any handle. + +## Rollout (staging) + +1. Terraform apply (new `options/staging/airdrop-bot` secret placeholder). +2. Provision the DB (one-time, infra/README.md convention): + `CREATE DATABASE engagement_staging; CREATE USER engagement_staging …`. +3. Fill the airdrop-bot secret with the new Discord application's public + key; register the slash commands (services/airdrop-bot/README.md). +4. Deploy `engagement-service` + `airdrop-bot` (+ rebuilt + `twitter-service`); point the Discord app's Interactions Endpoint URL at + `https:///staging/airdrop-bot/discord/interactions`. +5. Set `points.ambassadors` in engagement-service's config as the + ambassador roster firms up. diff --git a/rust-backend/Cargo.lock b/rust-backend/Cargo.lock index f9d4ff68..a4add77f 100644 --- a/rust-backend/Cargo.lock +++ b/rust-backend/Cargo.lock @@ -101,6 +101,26 @@ dependencies = [ "memchr", ] +[[package]] +name = "airdrop-bot" +version = "0.1.0" +dependencies = [ + "anyhow", + "axum 0.7.9", + "clap", + "cli-spec", + "ed25519-dalek", + "hex", + "observability", + "rand 0.8.6", + "reqwest", + "runtime-config", + "serde", + "serde_json", + "tokio", + "tracing", +] + [[package]] name = "aliasable" version = "0.1.3" @@ -3022,6 +3042,27 @@ version = "0.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c34f04666d835ff5d62e058c3995147c06f42fe86ff053337632bca83e42702d" +[[package]] +name = "engagement-service" +version = "0.1.0" +dependencies = [ + "anyhow", + "axum 0.7.9", + "chrono", + "clap", + "cli-spec", + "diesel", + "diesel_migrations", + "observability", + "r2d2", + "reqwest", + "runtime-config", + "serde", + "serde_json", + "tokio", + "tracing", +] + [[package]] name = "enum-compat-util" version = "0.1.0" diff --git a/rust-backend/Cargo.toml b/rust-backend/Cargo.toml index 00743425..c54c60f0 100644 --- a/rust-backend/Cargo.toml +++ b/rust-backend/Cargo.toml @@ -31,6 +31,8 @@ members = [ "services/oracle-service", "services/twitter-service", "services/social-bot", + "services/engagement-service", + "services/airdrop-bot", "tools/deployment-manager", "tools/exchange", "tools/writer", diff --git a/rust-backend/Dockerfile.airdrop-bot b/rust-backend/Dockerfile.airdrop-bot new file mode 100644 index 00000000..2ea4feee --- /dev/null +++ b/rust-backend/Dockerfile.airdrop-bot @@ -0,0 +1,30 @@ +# Multi-stage build for airdrop-bot. +# +# The builder runs under the target platform set by bake.hcl. Do NOT add +# `--platform=$BUILDPLATFORM` here — that would pin the builder to the GH +# runner's native arch and produce a binary that doesn't match the image +# manifest's arch. +FROM rust:1-bookworm AS builder +WORKDIR /src + +RUN apt-get update && apt-get install -y --no-install-recommends \ + pkg-config libssl-dev clang cmake protobuf-compiler git \ + && rm -rf /var/lib/apt/lists/* + +# Copy the whole workspace. The Sui git deps make smarter caching strategies +# fragile; one COPY is simpler and the layer rebuilds only when the workspace +# changes. +COPY . . +RUN cargo build --release -p airdrop-bot + +FROM debian:bookworm-slim +RUN apt-get update && apt-get install -y --no-install-recommends \ + ca-certificates libssl3 curl && rm -rf /var/lib/apt/lists/* +WORKDIR /app +COPY --from=builder /src/target/release/airdrop-bot /usr/local/bin/airdrop-bot +COPY services/airdrop-bot/config/ /app/config/ + +ENV APP_ENV=staging +ENTRYPOINT ["/bin/sh", "-c", "exec /usr/local/bin/airdrop-bot \ + --config /app/config/config.${APP_ENV}.toml \ + --secrets /run/secrets/airdrop-bot.toml"] diff --git a/rust-backend/Dockerfile.engagement-service b/rust-backend/Dockerfile.engagement-service new file mode 100644 index 00000000..592d01f6 --- /dev/null +++ b/rust-backend/Dockerfile.engagement-service @@ -0,0 +1,30 @@ +# Multi-stage build for engagement-service. +# +# The builder runs under the target platform set by bake.hcl. Do NOT add +# `--platform=$BUILDPLATFORM` here — that would pin the builder to the GH +# runner's native arch and produce a binary that doesn't match the image +# manifest's arch. +FROM rust:1-bookworm AS builder +WORKDIR /src + +RUN apt-get update && apt-get install -y --no-install-recommends \ + pkg-config libssl-dev clang cmake protobuf-compiler git \ + && rm -rf /var/lib/apt/lists/* + +# Copy the whole workspace. The Sui git deps make smarter caching strategies +# fragile; one COPY is simpler and the layer rebuilds only when the workspace +# changes. +COPY . . +RUN cargo build --release -p engagement-service + +FROM debian:bookworm-slim +# libpq5: Postgres runtime lib for the diesel connection. +RUN apt-get update && apt-get install -y --no-install-recommends \ + ca-certificates libssl3 libpq5 curl && rm -rf /var/lib/apt/lists/* +WORKDIR /app +COPY --from=builder /src/target/release/engagement-service /usr/local/bin/engagement-service +COPY services/engagement-service/config/ /app/config/ + +ENV APP_ENV=staging +ENTRYPOINT ["/bin/sh", "-c", "exec /usr/local/bin/engagement-service \ + --config /app/config/config.${APP_ENV}.toml"] diff --git a/rust-backend/deployment/affected.py b/rust-backend/deployment/affected.py index 509509d9..60037bd3 100755 --- a/rust-backend/deployment/affected.py +++ b/rust-backend/deployment/affected.py @@ -34,7 +34,7 @@ # Order here is the canonical "all services" list. Keep in sync with the # ALL_SERVICES array in deployment/ec2/deploy.sh. -ALL_SERVICES = ["indexer", "quoting-service", "mm-bot", "option-scheduler", "api-service", "token-info", "auth-service", "gas-station", "price-charting", "balance-monitor", "keeper", "oracle-service", "cctp-relay", "twitter-service", "social-bot"] +ALL_SERVICES = ["indexer", "quoting-service", "mm-bot", "option-scheduler", "api-service", "token-info", "auth-service", "gas-station", "price-charting", "balance-monitor", "keeper", "oracle-service", "cctp-relay", "twitter-service", "social-bot", "engagement-service", "airdrop-bot"] # Path globs that, when matched, force every service to rebuild + # redeploy. Catches lockfile churn, workspace-wide config, infra-side @@ -73,6 +73,8 @@ # balance-monitor : runtime-config, cli-spec, sui-tx, observability # twitter-service : runtime-config, cli-spec # social-bot : runtime-config, cli-spec +# engagement-service: runtime-config, cli-spec +# airdrop-bot : runtime-config, cli-spec # keeper : protocol-types, runtime-config, cli-spec, sui-tx, # pyth-client, pricing, token-info-client, # indexer-graphql, observability @@ -219,6 +221,20 @@ "rust-backend/crates/observability/**", "rust-backend/crates/cli-spec/**", ], + "engagement-service": [ + "rust-backend/services/engagement-service/**", + "rust-backend/Dockerfile.engagement-service", + "rust-backend/crates/runtime-config/**", + "rust-backend/crates/observability/**", + "rust-backend/crates/cli-spec/**", + ], + "airdrop-bot": [ + "rust-backend/services/airdrop-bot/**", + "rust-backend/Dockerfile.airdrop-bot", + "rust-backend/crates/runtime-config/**", + "rust-backend/crates/observability/**", + "rust-backend/crates/cli-spec/**", + ], "oracle-service": [ "rust-backend/services/oracle-service/**", "rust-backend/Dockerfile.oracle-service", diff --git a/rust-backend/deployment/bake.hcl b/rust-backend/deployment/bake.hcl index 6d454f0c..ee81c0cd 100644 --- a/rust-backend/deployment/bake.hcl +++ b/rust-backend/deployment/bake.hcl @@ -147,6 +147,22 @@ target "social-bot" { cache-to = [{ type = "gha", mode = "max", scope = "social-bot" }] } +target "engagement-service" { + inherits = ["_common"] + dockerfile = "Dockerfile.engagement-service" + tags = ["${ECR}/options/engagement-service:${IMAGE_TAG}"] + cache-from = [{ type = "gha", scope = "engagement-service" }] + cache-to = [{ type = "gha", mode = "max", scope = "engagement-service" }] +} + +target "airdrop-bot" { + inherits = ["_common"] + dockerfile = "Dockerfile.airdrop-bot" + tags = ["${ECR}/options/airdrop-bot:${IMAGE_TAG}"] + cache-from = [{ type = "gha", scope = "airdrop-bot" }] + cache-to = [{ type = "gha", mode = "max", scope = "airdrop-bot" }] +} + group "default" { - targets = ["indexer", "quoting-service", "mm-bot", "option-scheduler", "api-service", "token-info", "auth-service", "gas-station", "price-charting", "keeper", "balance-monitor", "oracle-service", "cctp-relay", "twitter-service", "social-bot"] + targets = ["indexer", "quoting-service", "mm-bot", "option-scheduler", "api-service", "token-info", "auth-service", "gas-station", "price-charting", "keeper", "balance-monitor", "oracle-service", "cctp-relay", "twitter-service", "social-bot", "engagement-service", "airdrop-bot"] } diff --git a/rust-backend/deployment/compose/docker-compose.staging.yml b/rust-backend/deployment/compose/docker-compose.staging.yml index 62f09253..1d774d20 100644 --- a/rust-backend/deployment/compose/docker-compose.staging.yml +++ b/rust-backend/deployment/compose/docker-compose.staging.yml @@ -252,6 +252,40 @@ services: restart: unless-stopped networks: [net] + # Engagement→airdrop-points tracker (staging-only, like twitter-service). + # Polls twitter-service for mentions of our account, persists per-tweet + # engagement counters in the shared RDS Postgres, and serves the airdrop + # leaderboard. Internal-only port 9017 (never proxied by nginx) — + # airdrop-bot is the public surface. + engagement-service: + image: ${ECR}/options/engagement-service:${ENGAGEMENT_SERVICE_TAG} + environment: + APP_ENV: staging + OTEL_EXPORTER_OTLP_ENDPOINT: ${OTEL_ENDPOINT:-} + DB_PASSWORD: ${DB_PASSWORD} + DB_HOST: ${DB_HOST} + RUST_LOG: info,engagement_service=debug + depends_on: [twitter-service] + restart: unless-stopped + networks: [net] + + # Discord bot for the engagement airdrop (staging-only). Deliberately a + # separate Discord application + container from social-bot. Read-only + # /leaderboard + /points commands served from engagement-service. Public + # port 9018 (proxied by nginx — Discord delivers signed webhooks). Reads + # its Discord public key from /run/secrets/airdrop-bot.toml. + airdrop-bot: + image: ${ECR}/options/airdrop-bot:${AIRDROP_BOT_TAG} + environment: + APP_ENV: staging + OTEL_EXPORTER_OTLP_ENDPOINT: ${OTEL_ENDPOINT:-} + RUST_LOG: info,airdrop_bot=debug + volumes: + - /opt/options/staging/secrets:/run/secrets:ro + depends_on: [engagement-service] + restart: unless-stopped + networks: [net] + # Single public entrypoint per env. Strips the /staging/{quoting,api,indexer,scheduler,mm-bot} # prefix before forwarding. depends_on is start-order only; nginx # tolerates backends being briefly unreachable on first boot. @@ -268,7 +302,7 @@ services: volumes: - /opt/options/staging/nginx/:/etc/nginx/options/:ro command: ["nginx", "-c", "/etc/nginx/options/nginx.conf", "-g", "daemon off;"] - depends_on: [quoting, api-service, indexer, option-scheduler, mm-bot, token-info, auth-service, gas-station, keeper, social-bot] + depends_on: [quoting, api-service, indexer, option-scheduler, mm-bot, token-info, auth-service, gas-station, keeper, social-bot, airdrop-bot] restart: unless-stopped networks: [net] diff --git a/rust-backend/deployment/ec2/deploy.sh b/rust-backend/deployment/ec2/deploy.sh index 8a788cc1..57be2587 100755 --- a/rust-backend/deployment/ec2/deploy.sh +++ b/rust-backend/deployment/ec2/deploy.sh @@ -50,7 +50,7 @@ COMPOSE_FILE="docker-compose.${ENV}.yml" # Canonical service set + their .env tag-variable names + the compose # service name (mostly identical to the cargo crate name, except # quoting-service is referenced as `quoting` in compose). -ALL_SERVICES=(indexer quoting-service mm-bot option-scheduler api-service token-info auth-service gas-station price-charting balance-monitor keeper oracle-service cctp-relay twitter-service social-bot) +ALL_SERVICES=(indexer quoting-service mm-bot option-scheduler api-service token-info auth-service gas-station price-charting balance-monitor keeper oracle-service cctp-relay twitter-service social-bot engagement-service airdrop-bot) tag_var_for() { case "$1" in @@ -69,6 +69,8 @@ tag_var_for() { cctp-relay) echo CCTP_RELAY_TAG ;; twitter-service) echo TWITTER_SERVICE_TAG ;; social-bot) echo SOCIAL_BOT_TAG ;; + engagement-service) echo ENGAGEMENT_SERVICE_TAG ;; + airdrop-bot) echo AIRDROP_BOT_TAG ;; *) return 1 ;; esac } @@ -89,6 +91,8 @@ compose_name_for() { cctp-relay) echo cctp-relay ;; twitter-service) echo twitter-service ;; social-bot) echo social-bot ;; + engagement-service) echo engagement-service ;; + airdrop-bot) echo airdrop-bot ;; *) return 1 ;; esac } @@ -242,6 +246,7 @@ health_path_for() { cctp-relay) echo "/$ENV/cctp/health" ;; keeper) echo "/$ENV/keeper/health" ;; social-bot) echo "/$ENV/social-bot/health" ;; + airdrop-bot) echo "/$ENV/airdrop-bot/health" ;; *) return 1 ;; esac } diff --git a/rust-backend/deployment/ec2/render-secrets.sh b/rust-backend/deployment/ec2/render-secrets.sh index 839ddbf7..cf14deb4 100755 --- a/rust-backend/deployment/ec2/render-secrets.sh +++ b/rust-backend/deployment/ec2/render-secrets.sh @@ -296,6 +296,26 @@ discord_public_key = "$DISCORD_KEY" EOF fi +# ---- airdrop-bot secret -> rendered TOML ----------------------------------- +# Discord public key for the airdrop bot's OWN application (webhook +# verification) — a separate Discord app from social-bot's. Staging-only — +# absent in envs without the service, silently skipped. +# +# NOTE: airdrop-bot IS health-gated by deploy.sh (via nginx). The service +# refuses to boot on missing/placeholder values, so fill this secret before +# the first deploy that includes it, or the deploy rolls back. +if AIRDROP_JSON=$(fetch airdrop-bot 2>/dev/null); then + AIRDROP_DISCORD_KEY=$(echo "$AIRDROP_JSON" | jq -r '.discord_public_key') + if [ -z "$AIRDROP_DISCORD_KEY" ] || [ "$AIRDROP_DISCORD_KEY" = "null" ]; then + echo "missing discord_public_key in options/$ENV/airdrop-bot" >&2 + exit 1 + fi + umask 077 + cat > "$DIR/airdrop-bot.toml" < standalone [sui] rpc_url toml -------------------- # indexer / price-charting / balance-monitor hold no signing key but still # build a SuiClient. They read only `[sui] rpc_url` from these files (mounted diff --git a/rust-backend/deployment/nginx/nginx.staging.conf b/rust-backend/deployment/nginx/nginx.staging.conf index b786b527..6f2fb4d2 100644 --- a/rust-backend/deployment/nginx/nginx.staging.conf +++ b/rust-backend/deployment/nginx/nginx.staging.conf @@ -160,6 +160,18 @@ http { proxy_set_header X-Forwarded-Proto $scheme; } + # airdrop-bot Discord interactions endpoint (separate application from + # social-bot's). Requests are signature-verified by the service itself + # (Ed25519). engagement-service (9017) is deliberately never routed + # here — it's internal-only. + location ~ ^/staging/airdrop-bot(?:/(?.*))?$ { + set $upstream "airdrop-bot:9018"; + proxy_pass http://$upstream/$tail$is_args$args; + proxy_set_header Host $host; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + } + # vault-keeper health/metrics surface (8086 — see runtime_config::health). location ~ ^/staging/keeper(?:/(?.*))?$ { set $upstream "keeper:8086"; diff --git a/rust-backend/infra/secrets.tf b/rust-backend/infra/secrets.tf index f6b123f8..cc1ba5c1 100644 --- a/rust-backend/infra/secrets.tf +++ b/rust-backend/infra/secrets.tf @@ -240,6 +240,31 @@ resource "aws_secretsmanager_secret_version" "social_bot_placeholder" { } } +# airdrop-bot secret — Discord public key for webhook verification. A +# SEPARATE Discord application from social-bot's (community-facing bot). +# Staging-only (like mm-bot). Placeholder shape; put the real value by hand +# after apply: +# aws secretsmanager put-secret-value --secret-id options/staging/airdrop-bot \ +# --secret-string '{"discord_public_key":"..."}' +resource "aws_secretsmanager_secret" "airdrop_bot" { + for_each = toset(["staging"]) + name = "options/${each.key}/airdrop-bot" + description = "airdrop-bot webhook verification key (JSON: discord_public_key)." + recovery_window_in_days = 7 +} + +resource "aws_secretsmanager_secret_version" "airdrop_bot_placeholder" { + for_each = aws_secretsmanager_secret.airdrop_bot + secret_id = each.value.id + secret_string = jsonencode({ + discord_public_key = "REPLACE_ME" + }) + lifecycle { + # Operator updates this by hand after apply; don't drift back. + ignore_changes = [secret_string] + } +} + # Shared Sui JSON-RPC endpoint (SO-270). One secret per env — every service # that talks to a Sui fullnode reads this single URL so we point the whole # fleet at our rate-limit-lifted RPC provider without duplicating the token diff --git a/rust-backend/services/airdrop-bot/Cargo.toml b/rust-backend/services/airdrop-bot/Cargo.toml new file mode 100644 index 00000000..57356fbd --- /dev/null +++ b/rust-backend/services/airdrop-bot/Cargo.toml @@ -0,0 +1,39 @@ +[package] +name = "airdrop-bot" +version.workspace = true +edition.workspace = true +license.workspace = true + +[lib] +path = "src/lib.rs" + +[[bin]] +name = "airdrop-bot" +path = "src/main.rs" + +[dependencies] +runtime-config = { workspace = true } +observability = { workspace = true, features = ["axum"] } +cli-spec = { workspace = true } + +clap = { workspace = true } + +tokio = { workspace = true } + +axum = { workspace = true } + +serde = { workspace = true } +serde_json = { workspace = true } + +reqwest = { workspace = true } + +# Discord interaction verification (Ed25519 over the raw request body). +hex = { workspace = true } +ed25519-dalek = { workspace = true } + +anyhow = { workspace = true } + +tracing = { workspace = true } + +[dev-dependencies] +rand = { workspace = true } diff --git a/rust-backend/services/airdrop-bot/README.md b/rust-backend/services/airdrop-bot/README.md new file mode 100644 index 00000000..99129719 --- /dev/null +++ b/rust-backend/services/airdrop-bot/README.md @@ -0,0 +1,65 @@ +# airdrop-bot + +Discord slash-command bot for the engagement airdrop. Deliberately a +**separate Discord application + deployment from social-bot** (the tweeting +bot): different audience (community server vs team), different blast radius, +no shared secrets. + +Commands are read-only, served from engagement-service — no allow list: + +``` +/leaderboard [count] — top authors by airdrop points (default 10, max 25) +/points — one twitter handle's points + rank +``` + +Discord delivers commands as signed HTTP webhooks (no gateway/socket +connection), so one axum server behind nginx serves everything: + +- `POST /discord/interactions` — verified with the application public key + (Ed25519). +- `GET /health` + +Public staging URL (nginx strips the prefix): + +``` +https:///staging/airdrop-bot/discord/interactions +``` + +## One-time platform setup + +1. Create a NEW application at discord.com/developers/applications (do not + reuse social-bot's). +2. General Information → copy the **Public Key** into the + `options/staging/airdrop-bot` AWS secret (`discord_public_key`). +3. Register the slash commands (once per application; needs the bot token): + + ```sh + curl -X PUT "https://discord.com/api/v10/applications//commands" \ + -H "Authorization: Bot " -H "Content-Type: application/json" \ + -d '[ + { + "name": "leaderboard", + "description": "Top authors by airdrop points", + "options": [ + {"type": 4, "name": "count", "description": "How many entries (max 25)", "required": false} + ] + }, + { + "name": "points", + "description": "Airdrop points for a twitter handle", + "options": [ + {"type": 3, "name": "handle", "description": "Twitter handle (with or without @)", "required": true} + ] + } + ]' + ``` + +4. Set General Information → **Interactions Endpoint URL** to + `https:///staging/airdrop-bot/discord/interactions`. Discord + verifies the endpoint with a signed PING on save, so deploy airdrop-bot + (with the real public key in the secret) first. +5. Install the app to the community server (Installation → Guild Install). + +The bot token is only needed for the one-time command registration above; +the service itself never uses it (interaction webhooks authenticate with the +interaction token Discord sends in each request). diff --git a/rust-backend/services/airdrop-bot/config/config.staging.toml b/rust-backend/services/airdrop-bot/config/config.staging.toml new file mode 100644 index 00000000..15040fe7 --- /dev/null +++ b/rust-backend/services/airdrop-bot/config/config.staging.toml @@ -0,0 +1,8 @@ +# airdrop-bot config — staging env. + +environment = "staging" + +# Proxied by nginx at /staging/airdrop-bot/ — Discord webhooks land here. +bind_addr = "0.0.0.0:9018" + +engagement_service_url = "http://engagement-service:9017" diff --git a/rust-backend/services/airdrop-bot/config/config.toml b/rust-backend/services/airdrop-bot/config/config.toml new file mode 100644 index 00000000..99d09d07 --- /dev/null +++ b/rust-backend/services/airdrop-bot/config/config.toml @@ -0,0 +1,12 @@ +# airdrop-bot config (local dev). +# +# Discord slash-command bot for the engagement airdrop (/leaderboard, +# /points). The Discord application public key comes from the secrets TOML +# (`--secrets`, default services/airdrop-bot/config/secrets.toml); copy +# secrets.example.toml to secrets.toml and fill it in. + +environment = "dev" + +bind_addr = "127.0.0.1:9018" + +engagement_service_url = "http://127.0.0.1:9017" diff --git a/rust-backend/services/airdrop-bot/config/secrets.example.toml b/rust-backend/services/airdrop-bot/config/secrets.example.toml new file mode 100644 index 00000000..09857aba --- /dev/null +++ b/rust-backend/services/airdrop-bot/config/secrets.example.toml @@ -0,0 +1,8 @@ +# airdrop-bot secrets. Copy to secrets.toml for local dev; deployed envs +# render this from AWS Secrets Manager (options//airdrop-bot) via +# render-secrets.sh. + +# Discord application public key (hex) — General Information page of THIS +# bot's application in the developer portal (a separate application from +# social-bot's). +discord_public_key = "REPLACE_ME" diff --git a/rust-backend/services/airdrop-bot/src/commands.rs b/rust-backend/services/airdrop-bot/src/commands.rs new file mode 100644 index 00000000..44e4746d --- /dev/null +++ b/rust-backend/services/airdrop-bot/src/commands.rs @@ -0,0 +1,127 @@ +//! The /leaderboard and /points commands: fetch from engagement-service and +//! format the chat reply. + +use std::sync::Arc; + +use tracing::warn; + +use crate::engagement_client::Entry; +use crate::state::AppState; + +const DEFAULT_LEADERBOARD_COUNT: usize = 10; +const MAX_LEADERBOARD_COUNT: usize = 25; + +fn fmt_entry_line(e: &Entry) -> String { + let star = if e.ambassador { " ⭐" } else { "" }; + format!( + "{}. @{} — {:.0} airdrop pts ({:.0} engagement pts, {} tweet{}){}", + e.rank, + e.handle, + e.airdrop_points, + e.engagement_points, + e.tweets, + if e.tweets == 1 { "" } else { "s" }, + star + ) +} + +pub fn format_leaderboard(entries: &[Entry]) -> String { + if entries.is_empty() { + return "No tracked engagement yet — mention us on X to get on the board!".to_string(); + } + let mut lines = vec!["🏆 **Airdrop leaderboard** (⭐ = ambassador)".to_string()]; + lines.extend(entries.iter().map(fmt_entry_line)); + lines.join("\n") +} + +pub fn format_points(handle: &str, entry: Option<&Entry>) -> String { + match entry { + Some(e) => format!( + "@{} is rank #{} with {:.0} airdrop pts{} — {:.0} engagement pts from {} tweet{} \ + ({} likes, {} retweets, {} replies, {} quotes).", + e.handle, + e.rank, + e.airdrop_points, + if e.ambassador { " ⭐ (ambassador)" } else { "" }, + e.engagement_points, + e.tweets, + if e.tweets == 1 { "" } else { "s" }, + e.likes, + e.retweets, + e.replies, + e.quotes, + ), + None => format!( + "No tracked engagement for `@{}` yet — tweets mentioning us start counting \ + within a few minutes of posting.", + handle.trim_start_matches('@') + ), + } +} + +/// Run /leaderboard and produce the user-facing message. +pub async fn run_leaderboard(state: &Arc, count: Option) -> String { + let count = count + .unwrap_or(DEFAULT_LEADERBOARD_COUNT) + .clamp(1, MAX_LEADERBOARD_COUNT); + match state.engagement.leaderboard(count).await { + Ok(entries) => format_leaderboard(&entries), + Err(e) => { + warn!(error = %format!("{e:#}"), "leaderboard fetch failed"); + "❌ Couldn't reach the engagement service — try again in a minute.".to_string() + } + } +} + +/// Run /points and produce the user-facing message. +pub async fn run_points(state: &Arc, handle: &str) -> String { + match state.engagement.points(handle).await { + Ok(entry) => format_points(handle, entry.as_ref()), + Err(e) => { + warn!(handle, error = %format!("{e:#}"), "points fetch failed"); + "❌ Couldn't reach the engagement service — try again in a minute.".to_string() + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn entry(rank: usize, handle: &str, ambassador: bool) -> Entry { + Entry { + rank, + handle: handle.to_string(), + ambassador, + tweets: 2, + likes: 10, + retweets: 1, + replies: 3, + quotes: 0, + engagement_points: 19.0, + airdrop_points: if ambassador { 285.0 } else { 190.0 }, + } + } + + #[test] + fn formats_leaderboard_with_ambassador_star() { + let out = format_leaderboard(&[entry(1, "amber", true), entry(2, "bob", false)]); + assert!(out.contains("1. @amber — 285 airdrop pts (19 engagement pts, 2 tweets) ⭐")); + assert!(out.contains("2. @bob — 190 airdrop pts")); + assert!(!out.contains("bob — 190 airdrop pts (19 engagement pts, 2 tweets) ⭐")); + } + + #[test] + fn formats_empty_leaderboard() { + assert!(format_leaderboard(&[]).contains("No tracked engagement yet")); + } + + #[test] + fn formats_points_and_unknown_handle() { + let e = entry(3, "amber", true); + let out = format_points("amber", Some(&e)); + assert!(out.contains("rank #3")); + assert!(out.contains("ambassador")); + assert!(format_points("@ghost", None).contains("`@ghost`")); + } +} diff --git a/rust-backend/services/airdrop-bot/src/config.rs b/rust-backend/services/airdrop-bot/src/config.rs new file mode 100644 index 00000000..84b9bb17 --- /dev/null +++ b/rust-backend/services/airdrop-bot/src/config.rs @@ -0,0 +1,25 @@ +use std::net::SocketAddr; +use std::path::Path; + +use anyhow::Result; +use runtime_config::config_load; +use serde::Deserialize; + +#[derive(Debug, Clone, Deserialize)] +pub struct Config { + /// Deployment environment: `dev` / `staging` / `prod`. Logging only. + pub environment: String, + + /// Bind address. Proxied by nginx at //airdrop-bot/ — Discord + /// delivers signed interaction webhooks here. + pub bind_addr: SocketAddr, + + /// engagement-service base URL (internal compose network). + pub engagement_service_url: String, +} + +impl Config { + pub fn load>(path: P) -> Result { + config_load::load_toml(path) + } +} diff --git a/rust-backend/services/airdrop-bot/src/discord.rs b/rust-backend/services/airdrop-bot/src/discord.rs new file mode 100644 index 00000000..ff045b8d --- /dev/null +++ b/rust-backend/services/airdrop-bot/src/discord.rs @@ -0,0 +1,195 @@ +//! Discord interactions endpoint (`POST /discord/interactions`). +//! +//! Same webhook shape as social-bot's: Discord signs every interaction with +//! the application's Ed25519 key over `` and requires the +//! PING handshake plus an ack within 3s. The handler defers, fetches from +//! engagement-service in a background task, and edits the deferred response +//! through the interaction webhook. + +use std::sync::Arc; + +use axum::body::Bytes; +use axum::extract::State; +use axum::http::{HeaderMap, StatusCode}; +use axum::response::{IntoResponse, Json, Response}; +use ed25519_dalek::{Signature, Verifier, VerifyingKey}; +use serde_json::{json, Value}; +use tracing::warn; + +use crate::commands; +use crate::state::AppState; + +const DISCORD_API: &str = "https://discord.com/api/v10"; + +// Interaction types / callback types, from Discord's API reference. +const PING: u64 = 1; +const APPLICATION_COMMAND: u64 = 2; +const PONG: u64 = 1; +const CHANNEL_MESSAGE: u64 = 4; +const DEFERRED_CHANNEL_MESSAGE: u64 = 5; +const EPHEMERAL: u64 = 1 << 6; + +pub async fn interactions( + State(state): State>, + headers: HeaderMap, + body: Bytes, +) -> Response { + if let Err(reason) = verify_signature(&state.discord_verify_key, &headers, &body) { + warn!(reason, "rejected discord request"); + return (StatusCode::UNAUTHORIZED, "invalid signature").into_response(); + } + + let Ok(interaction) = serde_json::from_slice::(&body) else { + return (StatusCode::BAD_REQUEST, "bad json").into_response(); + }; + + match interaction["type"].as_u64() { + Some(PING) => Json(json!({ "type": PONG })).into_response(), + Some(APPLICATION_COMMAND) => command(state, &interaction).await, + _ => (StatusCode::BAD_REQUEST, "unsupported interaction type").into_response(), + } +} + +async fn command(state: Arc, interaction: &Value) -> Response { + let option = |name: &str| -> Option { + interaction["data"]["options"] + .as_array() + .and_then(|opts| opts.iter().find(|o| o["name"].as_str() == Some(name))) + .map(|o| o["value"].clone()) + }; + + // Both commands are read-only, so there is no allow list. + enum Cmd { + Leaderboard { count: Option }, + Points { handle: String }, + } + let cmd = match interaction["data"]["name"].as_str() { + Some("leaderboard") => Cmd::Leaderboard { + count: option("count").and_then(|v| v.as_u64()).map(|n| n as usize), + }, + Some("points") => match option("handle").and_then(|v| v.as_str().map(str::to_string)) { + Some(handle) if !handle.trim().is_empty() => Cmd::Points { + handle: handle.trim().to_string(), + }, + _ => return ephemeral_message("Usage: /points "), + }, + _ => return ephemeral_message("Unknown command."), + }; + + let application_id = interaction["application_id"] + .as_str() + .unwrap_or_default() + .to_string(); + let token = interaction["token"].as_str().unwrap_or_default().to_string(); + + // Defer now; fetch + edit the deferred response from a background task. + let task_state = state.clone(); + tokio::spawn(async move { + let message = match cmd { + Cmd::Leaderboard { count } => commands::run_leaderboard(&task_state, count).await, + Cmd::Points { handle } => commands::run_points(&task_state, &handle).await, + }; + let url = format!("{DISCORD_API}/webhooks/{application_id}/{token}/messages/@original"); + let body = json!({ "content": message }); + let send = |headers| { + task_state + .http + .patch(&url) + .headers(headers) + .json(&body) + .send() + }; + match observability::client::instrumented("discord", "PATCH webhook", send).await { + Ok(resp) if !resp.status().is_success() => { + warn!(status = %resp.status(), "discord follow-up failed"); + } + Err(e) => warn!(error = %e, "discord follow-up failed"), + _ => {} + } + }); + + Json(json!({ "type": DEFERRED_CHANNEL_MESSAGE })).into_response() +} + +fn ephemeral_message(text: &str) -> Response { + Json(json!({ + "type": CHANNEL_MESSAGE, + "data": { "content": text, "flags": EPHEMERAL }, + })) + .into_response() +} + +/// Check `X-Signature-Ed25519` over ``. +fn verify_signature( + key: &VerifyingKey, + headers: &HeaderMap, + body: &[u8], +) -> Result<(), &'static str> { + let timestamp = headers + .get("x-signature-timestamp") + .and_then(|v| v.to_str().ok()) + .ok_or("missing timestamp header")?; + let sig_hex = headers + .get("x-signature-ed25519") + .and_then(|v| v.to_str().ok()) + .ok_or("missing signature header")?; + let sig_bytes: [u8; 64] = hex::decode(sig_hex) + .map_err(|_| "bad signature hex")? + .try_into() + .map_err(|_| "bad signature length")?; + let signature = Signature::from_bytes(&sig_bytes); + + let mut message = Vec::with_capacity(timestamp.len() + body.len()); + message.extend_from_slice(timestamp.as_bytes()); + message.extend_from_slice(body); + key.verify(&message, &signature) + .map_err(|_| "signature mismatch") +} + +/// Parse the application public key from the Discord developer portal (hex). +pub fn parse_public_key(hex_key: &str) -> anyhow::Result { + let bytes: [u8; 32] = hex::decode(hex_key.trim()) + .map_err(|e| anyhow::anyhow!("discord public key is not hex: {e}"))? + .try_into() + .map_err(|_| anyhow::anyhow!("discord public key must be 32 bytes"))?; + VerifyingKey::from_bytes(&bytes).map_err(|e| anyhow::anyhow!("bad discord public key: {e}")) +} + +#[cfg(test)] +mod tests { + use super::*; + use ed25519_dalek::{Signer, SigningKey}; + + fn signed_headers(key: &SigningKey, timestamp: &str, body: &[u8]) -> HeaderMap { + let mut message = timestamp.as_bytes().to_vec(); + message.extend_from_slice(body); + let sig = key.sign(&message); + let mut h = HeaderMap::new(); + h.insert("x-signature-timestamp", timestamp.parse().unwrap()); + h.insert( + "x-signature-ed25519", + hex::encode(sig.to_bytes()).parse().unwrap(), + ); + h + } + + #[test] + fn accepts_valid_signature_and_rejects_tampering() { + let signing = SigningKey::generate(&mut rand::rngs::OsRng); + let verifying = signing.verifying_key(); + let body = br#"{"type":1}"#; + + let headers = signed_headers(&signing, "1700000000", body); + assert_eq!(verify_signature(&verifying, &headers, body), Ok(())); + assert!(verify_signature(&verifying, &headers, br#"{"type":2}"#).is_err()); + } + + #[test] + fn parses_portal_hex_key() { + let signing = SigningKey::generate(&mut rand::rngs::OsRng); + let hex_key = hex::encode(signing.verifying_key().to_bytes()); + let parsed = parse_public_key(&hex_key).unwrap(); + assert_eq!(parsed, signing.verifying_key()); + assert!(parse_public_key("nothex").is_err()); + } +} diff --git a/rust-backend/services/airdrop-bot/src/engagement_client.rs b/rust-backend/services/airdrop-bot/src/engagement_client.rs new file mode 100644 index 00000000..10223921 --- /dev/null +++ b/rust-backend/services/airdrop-bot/src/engagement_client.rs @@ -0,0 +1,90 @@ +//! Client for engagement-service's internal HTTP API. + +use std::time::Duration; + +use anyhow::{Context, Result}; +use serde::Deserialize; + +const REQUEST_TIMEOUT: Duration = Duration::from_secs(10); + +pub struct EngagementClient { + http: reqwest::Client, + base_url: String, +} + +/// One leaderboard row (also the `GET /points/{handle}` payload). +#[derive(Debug, Clone, Deserialize)] +pub struct Entry { + pub rank: usize, + pub handle: String, + pub ambassador: bool, + pub tweets: i64, + pub likes: i64, + pub retweets: i64, + pub replies: i64, + pub quotes: i64, + pub engagement_points: f64, + pub airdrop_points: f64, +} + +#[derive(Debug, Deserialize)] +struct LeaderboardResp { + leaderboard: Vec, +} + +impl EngagementClient { + pub fn new(base_url: &str) -> Result { + let http = reqwest::Client::builder() + .timeout(REQUEST_TIMEOUT) + .build() + .context("building engagement-service http client")?; + Ok(Self { + http, + base_url: base_url.trim_end_matches('/').to_string(), + }) + } + + /// `GET /leaderboard?limit=N`. + pub async fn leaderboard(&self, limit: usize) -> Result> { + let url = format!("{}/leaderboard", self.base_url); + let resp = observability::client::instrumented("engagement-service", "GET /leaderboard", { + |headers| { + self.http + .get(&url) + .headers(headers) + .query(&[("limit", limit)]) + .send() + } + }) + .await + .context("fetching leaderboard from engagement-service")?; + let parsed: LeaderboardResp = resp + .error_for_status() + .context("engagement-service /leaderboard")? + .json() + .await + .context("parsing leaderboard")?; + Ok(parsed.leaderboard) + } + + /// `GET /points/{handle}` — `None` when the handle has no tracked + /// engagement (404). + pub async fn points(&self, handle: &str) -> Result> { + let url = format!("{}/points/{}", self.base_url, handle.trim_start_matches('@')); + let resp = observability::client::instrumented("engagement-service", "GET /points", { + |headers| self.http.get(&url).headers(headers).send() + }) + .await + .context("fetching points from engagement-service")?; + if resp.status() == reqwest::StatusCode::NOT_FOUND { + return Ok(None); + } + let entry: Entry = resp + .error_for_status() + .context("engagement-service /points")? + .json() + .await + .context("parsing points")?; + Ok(Some(entry)) + } +} diff --git a/rust-backend/services/airdrop-bot/src/lib.rs b/rust-backend/services/airdrop-bot/src/lib.rs new file mode 100644 index 00000000..6c406924 --- /dev/null +++ b/rust-backend/services/airdrop-bot/src/lib.rs @@ -0,0 +1,62 @@ +//! airdrop-bot. +//! +//! Discord slash-command bot for the engagement airdrop. Deliberately a +//! SEPARATE Discord application + deployment from social-bot (the tweeting +//! bot): different audience (community server vs team), different blast +//! radius, and no shared secrets. +//! +//! Commands are read-only, so there is no allow list — anyone in a server +//! the app is installed in can query: +//! +//! /leaderboard [count] — top authors by airdrop points +//! /points — one twitter handle's points + rank +//! +//! Discord delivers commands as signed HTTP webhooks (no gateway +//! connection), proxied by nginx at //airdrop-bot/: +//! - `POST /discord/interactions` — Ed25519-verified interactions endpoint. +//! - `GET /health` + +pub mod commands; +pub mod config; +pub mod discord; +pub mod engagement_client; +pub mod router; +pub mod secrets; +pub mod state; + +pub use config::Config; +pub use secrets::BotSecrets; +pub use state::AppState; + +use std::path::PathBuf; + +use clap::Parser; + +#[derive(Parser, Debug)] +#[command( + name = "airdrop-bot", + about = "Discord bot: airdrop leaderboard + per-handle points from engagement-service." +)] +pub struct Cli { + #[arg(short, long, default_value = "services/airdrop-bot/config/config.toml")] + pub config: PathBuf, + + /// Secrets TOML holding the Discord application public key. + /// No env-var fallback. + #[arg( + short = 's', + long, + default_value = "services/airdrop-bot/config/secrets.toml" + )] + pub secrets: PathBuf, +} + +cli_spec::define_program! { + id = "airdrop-bot", + cargo_pkg = "airdrop-bot", + working_dir = ".", + description = "Discord slash-command bot for the engagement airdrop: /leaderboard and \ + /points, served from engagement-service. Separate Discord application \ + from social-bot.", + cli = crate::Cli, +} diff --git a/rust-backend/services/airdrop-bot/src/main.rs b/rust-backend/services/airdrop-bot/src/main.rs new file mode 100644 index 00000000..9eb3eca3 --- /dev/null +++ b/rust-backend/services/airdrop-bot/src/main.rs @@ -0,0 +1,45 @@ +use std::sync::Arc; +use std::time::Duration; + +use anyhow::{Context, Result}; +use clap::Parser; +use tracing::info; + +use airdrop_bot::engagement_client::EngagementClient; +use airdrop_bot::{discord, router, AppState, BotSecrets, Cli, Config}; + +#[tokio::main] +async fn main() -> Result<()> { + let _obs = observability::init("airdrop-bot"); + + let cli = Cli::parse(); + let cfg_path = cli.config.to_string_lossy().into_owned(); + info!(cfg_path, "loading config"); + let cfg = Config::load(&cfg_path).with_context(|| format!("loading config from {cfg_path}"))?; + + let secrets = BotSecrets::load(&cli.secrets) + .with_context(|| format!("loading secrets {}", cli.secrets.display()))?; + + let discord_verify_key = discord::parse_public_key(&secrets.discord_public_key) + .context("parsing discord public key from secrets")?; + + let engagement = EngagementClient::new(&cfg.engagement_service_url)?; + let http = reqwest::Client::builder() + .timeout(Duration::from_secs(10)) + .build() + .context("building follow-up http client")?; + + info!( + environment = %cfg.environment, + engagement_service_url = %cfg.engagement_service_url, + "airdrop-bot starting" + ); + + let state = Arc::new(AppState { + engagement, + http, + discord_verify_key, + }); + + router::serve(cfg.bind_addr, state).await +} diff --git a/rust-backend/services/airdrop-bot/src/router.rs b/rust-backend/services/airdrop-bot/src/router.rs new file mode 100644 index 00000000..a39dc18e --- /dev/null +++ b/rust-backend/services/airdrop-bot/src/router.rs @@ -0,0 +1,33 @@ +//! axum HTTP server. Proxied by nginx at //airdrop-bot/ — Discord +//! delivers its signed interaction webhooks through it. + +use std::net::SocketAddr; +use std::sync::Arc; + +use anyhow::Result; +use axum::routing::{get, post}; +use axum::Router; +use tracing::info; + +use crate::discord; +use crate::state::AppState; + +async fn health() -> &'static str { + "ok" +} + +pub async fn serve(addr: SocketAddr, state: Arc) -> Result<()> { + let app = Router::new() + .route("/health", get(health)) + .route("/discord/interactions", post(discord::interactions)) + .with_state(state) + .merge(observability::middleware::metrics_route()) + .layer(axum::middleware::from_fn( + observability::middleware::http_obs, + )); + + let listener = tokio::net::TcpListener::bind(addr).await?; + info!(%addr, "airdrop-bot http listening"); + axum::serve(listener, app).await?; + Ok(()) +} diff --git a/rust-backend/services/airdrop-bot/src/secrets.rs b/rust-backend/services/airdrop-bot/src/secrets.rs new file mode 100644 index 00000000..c193a7ed --- /dev/null +++ b/rust-backend/services/airdrop-bot/src/secrets.rs @@ -0,0 +1,28 @@ +//! Bot credentials, loaded from the secrets TOML (rendered by +//! render-secrets.sh from AWS Secrets Manager in deployed envs). + +use std::path::Path; + +use anyhow::{ensure, Result}; +use runtime_config::config_load; +use serde::Deserialize; + +#[derive(Debug, Clone, Deserialize)] +pub struct BotSecrets { + /// Discord application public key (hex). Verifies the Ed25519 signature + /// on `POST /discord/interactions`. This is the AIRDROP bot's own + /// application — not social-bot's. + pub discord_public_key: String, +} + +impl BotSecrets { + pub fn load>(path: P) -> Result { + let secrets: Self = config_load::load_toml(path)?; + ensure!( + !secrets.discord_public_key.trim().is_empty() + && secrets.discord_public_key != "REPLACE_ME", + "airdrop-bot secrets: discord_public_key is empty or a placeholder" + ); + Ok(secrets) + } +} diff --git a/rust-backend/services/airdrop-bot/src/state.rs b/rust-backend/services/airdrop-bot/src/state.rs new file mode 100644 index 00000000..8ab0c084 --- /dev/null +++ b/rust-backend/services/airdrop-bot/src/state.rs @@ -0,0 +1,14 @@ +//! Shared application state. + +use ed25519_dalek::VerifyingKey; + +use crate::engagement_client::EngagementClient; + +pub struct AppState { + /// engagement-service read API. + pub engagement: EngagementClient, + /// Follow-up client for editing deferred Discord responses. + pub http: reqwest::Client, + /// Discord application public key (this bot's own application). + pub discord_verify_key: VerifyingKey, +} diff --git a/rust-backend/services/engagement-service/Cargo.toml b/rust-backend/services/engagement-service/Cargo.toml new file mode 100644 index 00000000..9d89ca02 --- /dev/null +++ b/rust-backend/services/engagement-service/Cargo.toml @@ -0,0 +1,40 @@ +[package] +name = "engagement-service" +version.workspace = true +edition.workspace = true +license.workspace = true + +[lib] +path = "src/lib.rs" + +[[bin]] +name = "engagement-service" +path = "src/main.rs" + +[dependencies] +runtime-config = { workspace = true } +observability = { workspace = true, features = ["axum"] } +cli-spec = { workspace = true } + +clap = { workspace = true } + +tokio = { workspace = true } + +axum = { workspace = true } + +serde = { workspace = true } +serde_json = { workspace = true } + +reqwest = { workspace = true } + +chrono = { workspace = true } + +# Postgres persistence (same diesel + r2d2 + embedded-migration shape as +# price-charting). +diesel = { workspace = true } +diesel_migrations = { workspace = true } +r2d2 = { workspace = true } + +anyhow = { workspace = true } + +tracing = { workspace = true } diff --git a/rust-backend/services/engagement-service/README.md b/rust-backend/services/engagement-service/README.md new file mode 100644 index 00000000..e8b198ac --- /dev/null +++ b/rust-backend/services/engagement-service/README.md @@ -0,0 +1,67 @@ +# engagement-service + +Tracks engagement on tweets that mention our account and converts it into +airdrop points. One service owns the whole pipeline (MVP): + +``` +twitter-service ──GET /mentions──▶ poller ──▶ Postgres (tracked_tweets) + ◀─GET /tweets/metrics──┘ │ + ▼ +airdrop-bot ◀──GET /leaderboard, /points/{handle}──┘ +``` + +- **Tracking** — every `poll_interval_secs` (default 300) the poller asks + twitter-service for new tweets mentioning `twitter_account` (Twitter + recent search: original tweets only, last 7 days) and upserts them with + their like/retweet/reply/quote counters. It also refreshes the stalest + counters among tweets younger than `refresh_max_age_hours` (default 168), + 100 per tick. +- **Airdrop points** — derived at read time from `[points]` config weights + (never stored), so re-tuning weights or the ambassador roster is an + ordinary config deploy with no backfill: + + ``` + engagement_points = likes*like_weight + replies*reply_weight + + retweets*retweet_weight + quotes*quote_weight + airdrop_points = engagement_points * airdrop_points_per_engagement_point + * (ambassador ? ambassador_multiplier : 1) + ``` + + Ambassadors are twitter handles in `points.ambassadors` (config, not + secrets). Everyone else is "general public" — same leaderboard, no + multiplier. +- **Leaderboard** — authors ranked by airdrop points. + +Internal-only (never proxied by nginx); airdrop-bot is the public surface. + +## Endpoints + +- `GET /health` +- `GET /leaderboard?limit=N` — ranked entries (default 10, max 100): + `{rank, handle, ambassador, tweets, likes, retweets, replies, quotes, + engagement_points, airdrop_points}` +- `GET /points/{handle}` — one handle's entry + rank (404 when untracked). + Handles are lowercased at ingest; lookups are case-insensitive and a + leading `@` is accepted. + +## Persistence + +Postgres via diesel with embedded migrations (same shape as +price-charting). One row per tweet with its latest counters — no snapshot +history in the MVP. One-time provisioning per env (same convention as the +other DB-backed services): + +```sql +CREATE DATABASE engagement_staging; +CREATE USER engagement_staging WITH PASSWORD ''; +GRANT ALL PRIVILEGES ON DATABASE engagement_staging TO engagement_staging; +``` + +## Not in the MVP (deliberately) + +- Per-tweet snapshot history / engagement trend charts. +- Splitting airdrop conversion into its own service — it's a pure function + in `src/points.rs`; split it out when the airdrop program needs its own + state (claims, epochs, on-chain distribution). +- Follower-count weighting, spam/bot filtering beyond `-is:retweet`, and + mention search past Twitter's 7-day recent-search window. diff --git a/rust-backend/services/engagement-service/config/config.staging.toml b/rust-backend/services/engagement-service/config/config.staging.toml new file mode 100644 index 00000000..6a905653 --- /dev/null +++ b/rust-backend/services/engagement-service/config/config.staging.toml @@ -0,0 +1,32 @@ +# engagement-service config — staging env. + +environment = "staging" + +# Internal-only: reachable on the compose `net` network (airdrop-bot), +# deliberately never proxied by nginx. +bind_addr = "0.0.0.0:9017" + +database_url = "postgresql://engagement_staging:${DB_PASSWORD}@${DB_HOST}:5432/engagement_staging" +db_pool_size = 4 + +twitter_service_url = "http://twitter-service:9014" +# twitter-service account name (= the @handle whose mentions accrue points). +twitter_account = "suioptions" + +poll_interval_secs = 300 +refresh_max_age_hours = 168 + +[points] +# engagement_points = likes*1 + replies*2 + retweets*3 + quotes*4 +like_weight = 1.0 +reply_weight = 2.0 +retweet_weight = 3.0 +quote_weight = 4.0 + +# airdrop_points = engagement_points * rate * (ambassador ? multiplier : 1) +airdrop_points_per_engagement_point = 10.0 +ambassador_multiplier = 1.5 +# Ambassador twitter handles (without @, case-insensitive). Config, not +# secrets — handles aren't sensitive and updating the roster is an ordinary +# config deploy. +ambassadors = [] diff --git a/rust-backend/services/engagement-service/config/config.toml b/rust-backend/services/engagement-service/config/config.toml new file mode 100644 index 00000000..e02183b2 --- /dev/null +++ b/rust-backend/services/engagement-service/config/config.toml @@ -0,0 +1,31 @@ +# engagement-service config (local dev). +# +# Tracks engagement on tweets mentioning our account (via twitter-service) +# and serves the airdrop leaderboard. Point database_url at any Postgres. + +environment = "dev" + +bind_addr = "127.0.0.1:9017" + +database_url = "${ENGAGEMENT_DATABASE_URL}" +db_pool_size = 4 + +twitter_service_url = "http://127.0.0.1:9014" +# twitter-service account name (= the @handle whose mentions accrue points). +twitter_account = "suioptions" + +poll_interval_secs = 300 +refresh_max_age_hours = 168 + +[points] +# engagement_points = likes*1 + replies*2 + retweets*3 + quotes*4 +like_weight = 1.0 +reply_weight = 2.0 +retweet_weight = 3.0 +quote_weight = 4.0 + +# airdrop_points = engagement_points * rate * (ambassador ? multiplier : 1) +airdrop_points_per_engagement_point = 10.0 +ambassador_multiplier = 1.5 +# Ambassador twitter handles (without @, case-insensitive). +ambassadors = [] diff --git a/rust-backend/services/engagement-service/src/config.rs b/rust-backend/services/engagement-service/src/config.rs new file mode 100644 index 00000000..389ddbae --- /dev/null +++ b/rust-backend/services/engagement-service/src/config.rs @@ -0,0 +1,60 @@ +//! Service config. Loaded via `runtime_config::config_load` so `${DB_HOST}` / +//! `${DB_PASSWORD}` expand from the environment at boot. + +use std::net::SocketAddr; +use std::path::Path; + +use anyhow::Result; +use runtime_config::config_load; +use serde::Deserialize; + +use crate::points::PointsConfig; + +fn default_db_pool_size() -> u32 { + 4 +} +fn default_poll_interval_secs() -> u64 { + 300 +} +fn default_refresh_max_age_hours() -> i64 { + 168 +} + +#[derive(Debug, Clone, Deserialize)] +pub struct Config { + /// Deployment environment: `dev` / `staging` / `prod`. Logging only. + pub environment: String, + + /// Bind address. Internal-only — reachable on the compose `net` network, + /// never proxied by nginx. + pub bind_addr: SocketAddr, + + /// Shared RDS Postgres, assembled from `${DB_HOST}` / `${DB_PASSWORD}`. + pub database_url: String, + #[serde(default = "default_db_pool_size")] + pub db_pool_size: u32, + + pub twitter_service_url: String, + /// twitter-service account name whose mentions accrue points. The name + /// doubles as the searched @handle (twitter-service secrets name + /// accounts by handle). + pub twitter_account: String, + + /// Seconds between poll ticks (mention search + metrics refresh). + #[serde(default = "default_poll_interval_secs")] + pub poll_interval_secs: u64, + /// Stop refreshing a tweet's counters once it is older than this — + /// engagement on week-old tweets has flattened, and recent search only + /// covers 7 days anyway. + #[serde(default = "default_refresh_max_age_hours")] + pub refresh_max_age_hours: i64, + + /// Engagement→airdrop-point conversion weights. + pub points: PointsConfig, +} + +impl Config { + pub fn load>(path: P) -> Result { + config_load::load_toml(path) + } +} diff --git a/rust-backend/services/engagement-service/src/db/migrations/000001_init/down.sql b/rust-backend/services/engagement-service/src/db/migrations/000001_init/down.sql new file mode 100644 index 00000000..d663622f --- /dev/null +++ b/rust-backend/services/engagement-service/src/db/migrations/000001_init/down.sql @@ -0,0 +1,2 @@ +DROP TABLE poll_cursor; +DROP TABLE tracked_tweets; diff --git a/rust-backend/services/engagement-service/src/db/migrations/000001_init/up.sql b/rust-backend/services/engagement-service/src/db/migrations/000001_init/up.sql new file mode 100644 index 00000000..dede25ca --- /dev/null +++ b/rust-backend/services/engagement-service/src/db/migrations/000001_init/up.sql @@ -0,0 +1,26 @@ +-- Tweets mentioning our account, with their latest engagement counters. +CREATE TABLE tracked_tweets ( + tweet_id TEXT PRIMARY KEY, + author_id TEXT NOT NULL, + author_handle TEXT NOT NULL, + text TEXT NOT NULL, + tweet_created_at TIMESTAMPTZ NOT NULL, + first_seen_at TIMESTAMPTZ NOT NULL, + metrics_updated_at TIMESTAMPTZ NOT NULL, + likes INT8 NOT NULL DEFAULT 0, + retweets INT8 NOT NULL DEFAULT 0, + replies INT8 NOT NULL DEFAULT 0, + quotes INT8 NOT NULL DEFAULT 0 +); + +-- Leaderboard groups by author; refresh scans by age/staleness. +CREATE INDEX tracked_tweets_author_handle_idx ON tracked_tweets (author_handle); +CREATE INDEX tracked_tweets_refresh_idx + ON tracked_tweets (tweet_created_at, metrics_updated_at); + +-- since_id cursor for the mention search (singleton row, id = 1). +CREATE TABLE poll_cursor ( + id INT2 PRIMARY KEY, + since_id TEXT NOT NULL, + updated_at TIMESTAMPTZ NOT NULL +); diff --git a/rust-backend/services/engagement-service/src/db/mod.rs b/rust-backend/services/engagement-service/src/db/mod.rs new file mode 100644 index 00000000..87703ff7 --- /dev/null +++ b/rust-backend/services/engagement-service/src/db/mod.rs @@ -0,0 +1,30 @@ +//! Postgres persistence. Same diesel + r2d2 + embedded-migration shape as +//! price-charting's `db` module. + +pub mod models; +pub mod repo; +pub mod schema; + +use anyhow::{Context, Result}; +use diesel::pg::PgConnection; +use diesel::r2d2::{ConnectionManager, Pool}; +use diesel_migrations::{embed_migrations, EmbeddedMigrations, MigrationHarness}; + +pub type DbPool = Pool>; + +pub const MIGRATIONS: EmbeddedMigrations = embed_migrations!("src/db/migrations"); + +pub fn establish_pool(database_url: &str, max_size: u32) -> Result { + let manager = ConnectionManager::::new(database_url); + Pool::builder() + .max_size(max_size) + .build(manager) + .context("building r2d2 pool for the engagement DB") +} + +pub fn run_migrations(pool: &DbPool) -> Result<()> { + let mut conn = pool.get().context("checking out connection for migrations")?; + conn.run_pending_migrations(MIGRATIONS) + .map_err(|e| anyhow::anyhow!("running migrations: {e}"))?; + Ok(()) +} diff --git a/rust-backend/services/engagement-service/src/db/models.rs b/rust-backend/services/engagement-service/src/db/models.rs new file mode 100644 index 00000000..3581db1c --- /dev/null +++ b/rust-backend/services/engagement-service/src/db/models.rs @@ -0,0 +1,30 @@ +//! Diesel row types. + +use chrono::{DateTime, Utc}; +use diesel::prelude::*; + +use super::schema::{poll_cursor, tracked_tweets}; + +#[derive(Insertable, Queryable, Debug, Clone)] +#[diesel(table_name = tracked_tweets)] +pub struct TweetRow { + pub tweet_id: String, + pub author_id: String, + pub author_handle: String, + pub text: String, + pub tweet_created_at: DateTime, + pub first_seen_at: DateTime, + pub metrics_updated_at: DateTime, + pub likes: i64, + pub retweets: i64, + pub replies: i64, + pub quotes: i64, +} + +#[derive(Insertable, Queryable, AsChangeset, Debug, Clone)] +#[diesel(table_name = poll_cursor)] +pub struct CursorRow { + pub id: i16, + pub since_id: String, + pub updated_at: DateTime, +} diff --git a/rust-backend/services/engagement-service/src/db/repo.rs b/rust-backend/services/engagement-service/src/db/repo.rs new file mode 100644 index 00000000..839c92b8 --- /dev/null +++ b/rust-backend/services/engagement-service/src/db/repo.rs @@ -0,0 +1,165 @@ +//! Repository over the engagement DB. + +use anyhow::{Context, Result}; +use chrono::Utc; +use diesel::pg::PgConnection; +use diesel::prelude::*; +use diesel::r2d2::{ConnectionManager, PooledConnection}; +use diesel::sql_types::{BigInt, Text}; + +use crate::points::Engagement; + +use super::models::{CursorRow, TweetRow}; +use super::schema::{poll_cursor, tracked_tweets}; +use super::DbPool; + +#[derive(Clone)] +pub struct Repo { + pool: std::sync::Arc, +} + +/// Per-author engagement totals across every tracked tweet. +#[derive(QueryableByName, Debug, Clone)] +pub struct AuthorTotals { + #[diesel(sql_type = Text)] + pub author_handle: String, + #[diesel(sql_type = BigInt)] + pub tweets: i64, + #[diesel(sql_type = BigInt)] + pub likes: i64, + #[diesel(sql_type = BigInt)] + pub retweets: i64, + #[diesel(sql_type = BigInt)] + pub replies: i64, + #[diesel(sql_type = BigInt)] + pub quotes: i64, +} + +impl AuthorTotals { + pub fn engagement(&self) -> Engagement { + Engagement { + likes: self.likes, + retweets: self.retweets, + replies: self.replies, + quotes: self.quotes, + } + } +} + +impl Repo { + pub fn new(pool: std::sync::Arc) -> Self { + Self { pool } + } + + fn conn(&self) -> Result>> { + self.pool.get().context("checking out DB connection") + } + + pub fn load_since_id(&self) -> Result> { + let mut conn = self.conn()?; + let row = poll_cursor::table + .find(1i16) + .first::(&mut conn) + .optional() + .context("loading poll_cursor")?; + Ok(row.map(|r| r.since_id)) + } + + /// Upsert a mention batch (replays refresh the counters) and advance the + /// since_id cursor in one transaction. Returns rows written. + pub fn upsert_mentions(&self, rows: &[TweetRow], newest_id: Option<&str>) -> Result { + let mut conn = self.conn()?; + conn.transaction::<_, anyhow::Error, _>(|conn| { + let mut written = 0; + for row in rows { + written += diesel::insert_into(tracked_tweets::table) + .values(row) + .on_conflict(tracked_tweets::tweet_id) + .do_update() + .set(( + tracked_tweets::likes.eq(row.likes), + tracked_tweets::retweets.eq(row.retweets), + tracked_tweets::replies.eq(row.replies), + tracked_tweets::quotes.eq(row.quotes), + tracked_tweets::metrics_updated_at.eq(row.metrics_updated_at), + )) + .execute(conn) + .context("upserting tracked_tweets")?; + } + if let Some(newest) = newest_id { + diesel::insert_into(poll_cursor::table) + .values(CursorRow { + id: 1, + since_id: newest.to_string(), + updated_at: Utc::now(), + }) + .on_conflict(poll_cursor::id) + .do_update() + .set(( + poll_cursor::since_id.eq(newest), + poll_cursor::updated_at.eq(Utc::now()), + )) + .execute(conn) + .context("advancing poll_cursor")?; + } + Ok(written) + }) + } + + /// Tweet ids still young enough to refresh, stalest counters first. + pub fn refresh_candidates(&self, max_age_hours: i64, limit: i64) -> Result> { + #[derive(QueryableByName)] + struct IdRow { + #[diesel(sql_type = Text)] + tweet_id: String, + } + + let mut conn = self.conn()?; + // make_interval only accepts int4 args, hence the explicit cast. + let rows = diesel::sql_query( + "SELECT tweet_id FROM tracked_tweets \ + WHERE tweet_created_at > now() - make_interval(hours => $1::int4) \ + ORDER BY metrics_updated_at ASC LIMIT $2", + ) + .bind::(max_age_hours) + .bind::(limit) + .load::(&mut conn) + .context("querying refresh candidates")?; + Ok(rows.into_iter().map(|r| r.tweet_id).collect()) + } + + /// Overwrite counters for refreshed tweets. + pub fn update_metrics(&self, updates: &[(String, Engagement)]) -> Result<()> { + let mut conn = self.conn()?; + let now = Utc::now(); + for (tweet_id, e) in updates { + diesel::update(tracked_tweets::table.find(tweet_id)) + .set(( + tracked_tweets::likes.eq(e.likes), + tracked_tweets::retweets.eq(e.retweets), + tracked_tweets::replies.eq(e.replies), + tracked_tweets::quotes.eq(e.quotes), + tracked_tweets::metrics_updated_at.eq(now), + )) + .execute(&mut conn) + .context("updating tweet metrics")?; + } + Ok(()) + } + + /// Engagement totals per author, across every tracked tweet. + pub fn author_totals(&self) -> Result> { + let mut conn = self.conn()?; + diesel::sql_query( + "SELECT author_handle, \ + count(*) AS tweets, \ + coalesce(sum(likes), 0)::int8 AS likes, \ + coalesce(sum(retweets), 0)::int8 AS retweets, \ + coalesce(sum(replies), 0)::int8 AS replies, \ + coalesce(sum(quotes), 0)::int8 AS quotes \ + FROM tracked_tweets GROUP BY author_handle", + ) + .load::(&mut conn) + .context("querying author totals") + } +} diff --git a/rust-backend/services/engagement-service/src/db/schema.rs b/rust-backend/services/engagement-service/src/db/schema.rs new file mode 100644 index 00000000..eeb66b7b --- /dev/null +++ b/rust-backend/services/engagement-service/src/db/schema.rs @@ -0,0 +1,32 @@ +//! Hand-written diesel schema; kept in sync with `migrations/`. + +// One row per tweet mentioning the account, carrying the LATEST engagement +// counters (the leaderboard needs totals, not history — snapshots can be +// added later if trend charts are wanted). +diesel::table! { + tracked_tweets (tweet_id) { + tweet_id -> Text, + author_id -> Text, + // Lowercased at ingest — twitter handles are case-insensitive. + author_handle -> Text, + text -> Text, + tweet_created_at -> Timestamptz, + first_seen_at -> Timestamptz, + metrics_updated_at -> Timestamptz, + likes -> Int8, + retweets -> Int8, + replies -> Int8, + quotes -> Int8, + } +} + +// since_id cursor for the mention search (singleton row, id = 1). +diesel::table! { + poll_cursor (id) { + id -> Int2, + since_id -> Text, + updated_at -> Timestamptz, + } +} + +diesel::allow_tables_to_appear_in_same_query!(tracked_tweets, poll_cursor); diff --git a/rust-backend/services/engagement-service/src/handlers.rs b/rust-backend/services/engagement-service/src/handlers.rs new file mode 100644 index 00000000..7ae0687a --- /dev/null +++ b/rust-backend/services/engagement-service/src/handlers.rs @@ -0,0 +1,168 @@ +//! HTTP handlers: [`health`], [`leaderboard`], [`points`]. + +use std::sync::Arc; + +use axum::extract::{Json, Path, Query, State}; +use axum::http::StatusCode; +use serde::{Deserialize, Serialize}; +use tracing::warn; + +use crate::db::repo::AuthorTotals; +use crate::points::PointsConfig; +use crate::state::AppState; + +type ApiError = (StatusCode, String); + +pub async fn health() -> &'static str { + "ok" +} + +/// One leaderboard row (also the `GET /points/{handle}` payload). +#[derive(Serialize, Debug, Clone)] +pub struct LeaderboardEntry { + pub rank: usize, + pub handle: String, + pub ambassador: bool, + pub tweets: i64, + pub likes: i64, + pub retweets: i64, + pub replies: i64, + pub quotes: i64, + pub engagement_points: f64, + pub airdrop_points: f64, +} + +fn ranked_entries(points: &PointsConfig, totals: Vec) -> Vec { + let mut entries: Vec = totals + .into_iter() + .map(|t| { + let engagement_points = points.engagement_points(t.engagement()); + LeaderboardEntry { + rank: 0, + ambassador: points.is_ambassador(&t.author_handle), + airdrop_points: points.airdrop_points(&t.author_handle, engagement_points), + engagement_points, + tweets: t.tweets, + likes: t.likes, + retweets: t.retweets, + replies: t.replies, + quotes: t.quotes, + handle: t.author_handle, + } + }) + .collect(); + entries.sort_by(|a, b| { + b.airdrop_points + .total_cmp(&a.airdrop_points) + .then_with(|| a.handle.cmp(&b.handle)) + }); + for (i, e) in entries.iter_mut().enumerate() { + e.rank = i + 1; + } + entries +} + +fn load_entries(state: &AppState) -> Result, ApiError> { + let totals = state.repo.author_totals().map_err(|e| { + warn!(error = %format!("{e:#}"), "loading author totals failed"); + (StatusCode::INTERNAL_SERVER_ERROR, format!("{e:#}")) + })?; + Ok(ranked_entries(&state.cfg.points, totals)) +} + +fn default_limit() -> usize { + 10 +} + +#[derive(Deserialize)] +pub struct LeaderboardQuery { + #[serde(default = "default_limit")] + pub limit: usize, +} + +#[derive(Serialize)] +pub struct LeaderboardResp { + pub leaderboard: Vec, +} + +/// `GET /leaderboard?limit=N` — authors ranked by airdrop points. +pub async fn leaderboard( + State(s): State>, + Query(q): Query, +) -> Result, ApiError> { + let mut entries = load_entries(&s)?; + entries.truncate(q.limit.clamp(1, 100)); + Ok(Json(LeaderboardResp { + leaderboard: entries, + })) +} + +/// `GET /points/{handle}` — one author's totals, points and rank. +pub async fn points( + State(s): State>, + Path(handle): Path, +) -> Result, ApiError> { + let handle = handle.trim_start_matches('@').to_lowercase(); + let entries = load_entries(&s)?; + entries + .into_iter() + .find(|e| e.handle == handle) + .map(Json) + .ok_or_else(|| { + ( + StatusCode::NOT_FOUND, + format!("no tracked engagement for `{handle}`"), + ) + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn points_cfg(ambassadors: Vec) -> PointsConfig { + PointsConfig { + like_weight: 1.0, + reply_weight: 2.0, + retweet_weight: 3.0, + quote_weight: 4.0, + airdrop_points_per_engagement_point: 10.0, + ambassador_multiplier: 1.5, + ambassadors, + } + } + + fn totals(handle: &str, likes: i64) -> AuthorTotals { + AuthorTotals { + author_handle: handle.to_string(), + tweets: 1, + likes, + retweets: 0, + replies: 0, + quotes: 0, + } + } + + #[test] + fn ranks_by_airdrop_points_with_ambassador_boost() { + let cfg = points_cfg(vec!["amber".to_string()]); + // amber: 10 likes * 1 * 10 * 1.5 = 150; bob: 12 likes * 1 * 10 = 120. + let entries = ranked_entries(&cfg, vec![totals("bob", 12), totals("amber", 10)]); + assert_eq!(entries[0].handle, "amber"); + assert_eq!(entries[0].rank, 1); + assert!(entries[0].ambassador); + assert_eq!(entries[0].airdrop_points, 150.0); + assert_eq!(entries[1].handle, "bob"); + assert_eq!(entries[1].rank, 2); + assert!(!entries[1].ambassador); + assert_eq!(entries[1].airdrop_points, 120.0); + } + + #[test] + fn ties_break_alphabetically() { + let cfg = points_cfg(vec![]); + let entries = ranked_entries(&cfg, vec![totals("zed", 5), totals("ana", 5)]); + assert_eq!(entries[0].handle, "ana"); + assert_eq!(entries[1].handle, "zed"); + } +} diff --git a/rust-backend/services/engagement-service/src/lib.rs b/rust-backend/services/engagement-service/src/lib.rs new file mode 100644 index 00000000..ab5a0880 --- /dev/null +++ b/rust-backend/services/engagement-service/src/lib.rs @@ -0,0 +1,59 @@ +//! engagement-service. +//! +//! Tracks engagement on tweets that mention our account and converts it into +//! airdrop points (MVP: one service owns tracking, point conversion and the +//! leaderboard; the conversion lives in its own module so it can be split +//! out later if the airdrop program outgrows this). +//! +//! A poll loop asks twitter-service for new mentions of the configured +//! account and for refreshed engagement counters on tweets it already knows, +//! and persists both in Postgres. Points are derived at read time from +//! config weights, so re-tuning weights never needs a backfill. +//! +//! Internal-only: the bind port is reachable on the compose `net` network +//! (e.g. by airdrop-bot) and is deliberately never proxied by nginx. +//! +//! Endpoints: +//! - `GET /health` +//! - `GET /leaderboard?limit=N` — authors ranked by airdrop points. +//! - `GET /points/{handle}` — one author's totals, points and rank. + +pub mod config; +pub mod db; +pub mod handlers; +pub mod points; +pub mod poller; +pub mod router; +pub mod state; +pub mod twitter_client; + +pub use config::Config; +pub use state::AppState; + +use std::path::PathBuf; + +use clap::Parser; + +#[derive(Parser, Debug)] +#[command( + name = "engagement-service", + about = "Tracks tweet engagement for mentions of our account and serves the airdrop leaderboard." +)] +pub struct Cli { + #[arg( + short, + long, + default_value = "services/engagement-service/config/config.toml" + )] + pub config: PathBuf, +} + +cli_spec::define_program! { + id = "engagement-service", + cargo_pkg = "engagement-service", + working_dir = ".", + description = "Engagement + airdrop-points service. Polls twitter-service for mentions of \ + our account, persists per-tweet engagement counters, converts them into \ + airdrop points and serves the leaderboard.", + cli = crate::Cli, +} diff --git a/rust-backend/services/engagement-service/src/main.rs b/rust-backend/services/engagement-service/src/main.rs new file mode 100644 index 00000000..990c7bcf --- /dev/null +++ b/rust-backend/services/engagement-service/src/main.rs @@ -0,0 +1,41 @@ +use std::sync::Arc; + +use anyhow::{Context, Result}; +use clap::Parser; +use tracing::info; + +use engagement_service::db::{establish_pool, repo::Repo, run_migrations}; +use engagement_service::twitter_client::TwitterServiceClient; +use engagement_service::{poller, router, AppState, Cli, Config}; + +#[tokio::main] +async fn main() -> Result<()> { + let _obs = observability::init("engagement-service"); + + let cli = Cli::parse(); + let cfg_path = cli.config.to_string_lossy().into_owned(); + info!(cfg_path, "loading config"); + let cfg = Config::load(&cfg_path).with_context(|| format!("loading config from {cfg_path}"))?; + + let pool = Arc::new(establish_pool(&cfg.database_url, cfg.db_pool_size)?); + run_migrations(&pool).context("running engagement DB migrations")?; + let repo = Repo::new(pool); + info!(pool_size = cfg.db_pool_size, "engagement DB ready (migrations applied)"); + + let twitter = TwitterServiceClient::new(&cfg.twitter_service_url)?; + + info!( + environment = %cfg.environment, + twitter_service_url = %cfg.twitter_service_url, + account = %cfg.twitter_account, + poll_interval_secs = cfg.poll_interval_secs, + ambassadors = cfg.points.ambassadors.len(), + "engagement-service starting" + ); + + let state = Arc::new(AppState { repo, twitter, cfg }); + poller::spawn(Arc::clone(&state)); + + let bind_addr = state.cfg.bind_addr; + router::serve(bind_addr, state).await +} diff --git a/rust-backend/services/engagement-service/src/points.rs b/rust-backend/services/engagement-service/src/points.rs new file mode 100644 index 00000000..fa9ec8d0 --- /dev/null +++ b/rust-backend/services/engagement-service/src/points.rs @@ -0,0 +1,122 @@ +//! Engagement→airdrop-point conversion. Pure functions over config weights — +//! points are derived at read time, never stored, so re-tuning weights takes +//! effect immediately without a backfill. + +use serde::Deserialize; + +fn default_like_weight() -> f64 { + 1.0 +} +fn default_reply_weight() -> f64 { + 2.0 +} +fn default_retweet_weight() -> f64 { + 3.0 +} +fn default_quote_weight() -> f64 { + 4.0 +} +fn default_airdrop_rate() -> f64 { + 10.0 +} +fn default_ambassador_multiplier() -> f64 { + 1.5 +} + +#[derive(Debug, Clone, Deserialize)] +pub struct PointsConfig { + #[serde(default = "default_like_weight")] + pub like_weight: f64, + #[serde(default = "default_reply_weight")] + pub reply_weight: f64, + #[serde(default = "default_retweet_weight")] + pub retweet_weight: f64, + #[serde(default = "default_quote_weight")] + pub quote_weight: f64, + + /// Airdrop points granted per engagement point. + #[serde(default = "default_airdrop_rate")] + pub airdrop_points_per_engagement_point: f64, + /// Ambassadors' airdrop points are multiplied by this. + #[serde(default = "default_ambassador_multiplier")] + pub ambassador_multiplier: f64, + /// Ambassador twitter handles (without `@`, case-insensitive). + #[serde(default)] + pub ambassadors: Vec, +} + +/// Summed engagement counters (per tweet or per author). +#[derive(Debug, Clone, Copy, Default)] +pub struct Engagement { + pub likes: i64, + pub retweets: i64, + pub replies: i64, + pub quotes: i64, +} + +impl PointsConfig { + pub fn is_ambassador(&self, handle: &str) -> bool { + self.ambassadors.iter().any(|a| a.eq_ignore_ascii_case(handle)) + } + + pub fn engagement_points(&self, e: Engagement) -> f64 { + e.likes as f64 * self.like_weight + + e.replies as f64 * self.reply_weight + + e.retweets as f64 * self.retweet_weight + + e.quotes as f64 * self.quote_weight + } + + /// Engagement points → airdrop points, with the ambassador multiplier. + pub fn airdrop_points(&self, handle: &str, engagement_points: f64) -> f64 { + let multiplier = if self.is_ambassador(handle) { + self.ambassador_multiplier + } else { + 1.0 + }; + engagement_points * self.airdrop_points_per_engagement_point * multiplier + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn cfg() -> PointsConfig { + PointsConfig { + like_weight: 1.0, + reply_weight: 2.0, + retweet_weight: 3.0, + quote_weight: 4.0, + airdrop_points_per_engagement_point: 10.0, + ambassador_multiplier: 1.5, + ambassadors: vec!["Alice".to_string()], + } + } + + #[test] + fn weights_engagement() { + let e = Engagement { + likes: 10, + replies: 3, + retweets: 2, + quotes: 1, + }; + // 10*1 + 3*2 + 2*3 + 1*4 = 26 + assert_eq!(cfg().engagement_points(e), 26.0); + } + + #[test] + fn ambassador_matching_is_case_insensitive() { + let c = cfg(); + assert!(c.is_ambassador("alice")); + assert!(c.is_ambassador("ALICE")); + assert!(!c.is_ambassador("bob")); + } + + #[test] + fn converts_to_airdrop_points_with_multiplier() { + let c = cfg(); + assert_eq!(c.airdrop_points("bob", 26.0), 260.0); + assert_eq!(c.airdrop_points("alice", 26.0), 390.0); + } +} diff --git a/rust-backend/services/engagement-service/src/poller.rs b/rust-backend/services/engagement-service/src/poller.rs new file mode 100644 index 00000000..6fe0d441 --- /dev/null +++ b/rust-backend/services/engagement-service/src/poller.rs @@ -0,0 +1,109 @@ +//! Poll loop: ingest new mentions, refresh counters on known tweets. + +use std::sync::Arc; +use std::time::Duration; + +use anyhow::{Context, Result}; +use chrono::{DateTime, Utc}; +use tracing::{error, info, warn}; + +use crate::db::models::TweetRow; +use crate::points::Engagement; +use crate::state::AppState; + +/// Tweets per metrics-refresh call (Twitter's `GET /2/tweets` id cap). +const REFRESH_BATCH: i64 = 100; + +pub fn spawn(state: Arc) { + tokio::spawn(async move { + let mut interval = + tokio::time::interval(Duration::from_secs(state.cfg.poll_interval_secs.max(30))); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + loop { + interval.tick().await; + if let Err(e) = tick(&state).await { + // Grouped Grafana alert: engagement silently not accruing is + // the failure mode this service exists to avoid. + error!( + alert_id = "engagement-poll-failed", + error = %format!("{e:#}"), + "engagement poll tick failed" + ); + } + } + }); +} + +async fn tick(state: &AppState) -> Result<()> { + let account = &state.cfg.twitter_account; + + // New mentions since the stored cursor. + let since_id = state.repo.load_since_id()?; + let page = state + .twitter + .mentions(account, since_id.as_deref()) + .await + .context("fetching mentions")?; + let now = Utc::now(); + let rows: Vec = page + .mentions + .into_iter() + .filter_map(|m| { + let created = DateTime::parse_from_rfc3339(&m.created_at) + .map(|t| t.with_timezone(&Utc)) + .ok(); + if created.is_none() || m.author_handle.is_empty() { + warn!(tweet_id = %m.tweet_id, "skipping mention with missing author/timestamp"); + return None; + } + Some(TweetRow { + tweet_id: m.tweet_id, + author_id: m.author_id, + // Handles are case-insensitive; normalize so grouping and + // ambassador matching never split on case. + author_handle: m.author_handle.to_lowercase(), + text: m.text, + tweet_created_at: created.unwrap(), + first_seen_at: now, + metrics_updated_at: now, + likes: m.likes, + retweets: m.retweets, + replies: m.replies, + quotes: m.quotes, + }) + }) + .collect(); + let ingested = state.repo.upsert_mentions(&rows, page.newest_id.as_deref())?; + + // Refresh the stalest counters among tweets still young enough to move. + let ids = state + .repo + .refresh_candidates(state.cfg.refresh_max_age_hours, REFRESH_BATCH)?; + let mut refreshed = 0; + if !ids.is_empty() { + let metrics = state + .twitter + .tweets_metrics(account, &ids) + .await + .context("refreshing tweet metrics")?; + let updates: Vec<(String, Engagement)> = metrics + .into_iter() + .map(|m| { + ( + m.tweet_id, + Engagement { + likes: m.likes, + retweets: m.retweets, + replies: m.replies, + quotes: m.quotes, + }, + ) + }) + .collect(); + refreshed = updates.len(); + state.repo.update_metrics(&updates)?; + } + + info!(ingested, refreshed, "engagement poll tick"); + Ok(()) +} diff --git a/rust-backend/services/engagement-service/src/router.rs b/rust-backend/services/engagement-service/src/router.rs new file mode 100644 index 00000000..96efd7f9 --- /dev/null +++ b/rust-backend/services/engagement-service/src/router.rs @@ -0,0 +1,29 @@ +//! axum HTTP server (internal-only — never proxied by nginx). + +use std::net::SocketAddr; +use std::sync::Arc; + +use anyhow::Result; +use axum::routing::get; +use axum::Router; +use tracing::info; + +use crate::handlers; +use crate::state::AppState; + +pub async fn serve(addr: SocketAddr, state: Arc) -> Result<()> { + let app = Router::new() + .route("/health", get(handlers::health)) + .route("/leaderboard", get(handlers::leaderboard)) + .route("/points/:handle", get(handlers::points)) + .with_state(state) + .merge(observability::middleware::metrics_route()) + .layer(axum::middleware::from_fn( + observability::middleware::http_obs, + )); + + let listener = tokio::net::TcpListener::bind(addr).await?; + info!(%addr, "engagement-service http listening"); + axum::serve(listener, app).await?; + Ok(()) +} diff --git a/rust-backend/services/engagement-service/src/state.rs b/rust-backend/services/engagement-service/src/state.rs new file mode 100644 index 00000000..a1d9d9ba --- /dev/null +++ b/rust-backend/services/engagement-service/src/state.rs @@ -0,0 +1,11 @@ +//! Shared application state. + +use crate::config::Config; +use crate::db::repo::Repo; +use crate::twitter_client::TwitterServiceClient; + +pub struct AppState { + pub repo: Repo, + pub twitter: TwitterServiceClient, + pub cfg: Config, +} diff --git a/rust-backend/services/engagement-service/src/twitter_client.rs b/rust-backend/services/engagement-service/src/twitter_client.rs new file mode 100644 index 00000000..8fe016fe --- /dev/null +++ b/rust-backend/services/engagement-service/src/twitter_client.rs @@ -0,0 +1,99 @@ +//! Client for twitter-service's internal read API (mentions + metrics). + +use std::time::Duration; + +use anyhow::{Context, Result}; +use serde::Deserialize; + +const REQUEST_TIMEOUT: Duration = Duration::from_secs(15); + +pub struct TwitterServiceClient { + http: reqwest::Client, + base_url: String, +} + +/// One tweet mentioning the account, as `GET /mentions` returns it. +#[derive(Debug, Deserialize)] +pub struct Mention { + pub tweet_id: String, + pub author_id: String, + pub author_handle: String, + pub text: String, + /// RFC 3339. + pub created_at: String, + pub likes: i64, + pub retweets: i64, + pub replies: i64, + pub quotes: i64, +} + +#[derive(Debug, Deserialize)] +pub struct MentionsPage { + pub newest_id: Option, + pub mentions: Vec, +} + +#[derive(Debug, Deserialize)] +pub struct TweetMetrics { + pub tweet_id: String, + pub likes: i64, + pub retweets: i64, + pub replies: i64, + pub quotes: i64, +} + +#[derive(Debug, Deserialize)] +struct MetricsResp { + metrics: Vec, +} + +impl TwitterServiceClient { + pub fn new(base_url: &str) -> Result { + let http = reqwest::Client::builder() + .timeout(REQUEST_TIMEOUT) + .build() + .context("building twitter-service http client")?; + Ok(Self { + http, + base_url: base_url.trim_end_matches('/').to_string(), + }) + } + + /// `GET /mentions` — recent tweets mentioning `@account`. + pub async fn mentions(&self, account: &str, since_id: Option<&str>) -> Result { + let url = format!("{}/mentions", self.base_url); + let mut query = vec![("account", account.to_string())]; + if let Some(id) = since_id { + query.push(("since_id", id.to_string())); + } + let resp = observability::client::instrumented("twitter-service", "GET /mentions", { + |headers| self.http.get(&url).headers(headers).query(&query).send() + }) + .await + .context("fetching mentions from twitter-service")?; + resp.error_for_status() + .context("twitter-service /mentions")? + .json() + .await + .context("parsing mentions") + } + + /// `GET /tweets/metrics` — refreshed counters for up to 100 known tweets. + pub async fn tweets_metrics(&self, account: &str, ids: &[String]) -> Result> { + let url = format!("{}/tweets/metrics", self.base_url); + let query = [("account", account.to_string()), ("ids", ids.join(","))]; + let resp = + observability::client::instrumented("twitter-service", "GET /tweets/metrics", { + |headers| self.http.get(&url).headers(headers).query(&query).send() + }) + .await + .context("fetching tweet metrics from twitter-service")?; + let parsed: MetricsResp = resp + .error_for_status() + .context("twitter-service /tweets/metrics")? + .json() + .await + .context("parsing tweet metrics")?; + Ok(parsed.metrics) + } +} diff --git a/rust-backend/services/twitter-service/src/handlers.rs b/rust-backend/services/twitter-service/src/handlers.rs index f9c593e6..e5f75cb4 100644 --- a/rust-backend/services/twitter-service/src/handlers.rs +++ b/rust-backend/services/twitter-service/src/handlers.rs @@ -1,13 +1,16 @@ -//! HTTP handlers: [`health`], [`accounts`], [`post_tweet`]. +//! HTTP handlers: [`health`], [`accounts`], [`post_tweet`], [`mentions`], +//! [`tweets_metrics`]. use std::sync::Arc; -use axum::extract::{Json, State}; +use axum::extract::{Json, Query, State}; use axum::http::StatusCode; use serde::{Deserialize, Serialize}; -use tracing::{error, info}; +use tracing::{error, info, warn}; +use crate::secrets::TwitterAccount; use crate::state::AppState; +use crate::twitter::PublicMetrics; type ApiError = (StatusCode, String); @@ -35,17 +38,21 @@ pub struct PostTweetResp { pub text: String, } +fn account_creds<'a>(s: &'a AppState, account: &str) -> Result<&'a TwitterAccount, ApiError> { + s.accounts.get(account).ok_or_else(|| { + ( + StatusCode::NOT_FOUND, + format!("unknown account `{account}`"), + ) + }) +} + /// `POST /tweets` — post a tweet from the named account. pub async fn post_tweet( State(s): State>, Json(req): Json, ) -> Result, ApiError> { - let creds = s.accounts.get(&req.account).ok_or_else(|| { - ( - StatusCode::NOT_FOUND, - format!("unknown account `{}`", req.account), - ) - })?; + let creds = account_creds(&s, &req.account)?; if req.text.trim().is_empty() { return Err((StatusCode::BAD_REQUEST, "tweet text is empty".to_string())); } @@ -73,3 +80,143 @@ pub async fn post_tweet( } } } + +/// Engagement counters flattened to friendly names (consumed by +/// engagement-service). +#[derive(Serialize)] +pub struct MetricsDto { + pub likes: i64, + pub retweets: i64, + pub replies: i64, + pub quotes: i64, +} + +impl From for MetricsDto { + fn from(m: PublicMetrics) -> Self { + Self { + likes: m.like_count, + retweets: m.retweet_count, + replies: m.reply_count, + quotes: m.quote_count, + } + } +} + +#[derive(Deserialize)] +pub struct MentionsQuery { + /// Account name (= handle) whose mentions to search. + pub account: String, + /// Only return tweets newer than this id (the previous page's + /// `newest_id`). + pub since_id: Option, +} + +#[derive(Serialize)] +pub struct MentionDto { + pub tweet_id: String, + pub author_id: String, + pub author_handle: String, + pub text: String, + /// RFC 3339, as Twitter returns it. + pub created_at: String, + #[serde(flatten)] + pub metrics: MetricsDto, +} + +#[derive(Serialize)] +pub struct MentionsResp { + pub account: String, + /// `since_id` for the next poll. Absent when nothing matched. + pub newest_id: Option, + pub mentions: Vec, +} + +/// `GET /mentions?account=[&since_id=]` — recent (≤7 days) +/// original tweets mentioning `@account`. +pub async fn mentions( + State(s): State>, + Query(q): Query, +) -> Result, ApiError> { + let creds = account_creds(&s, &q.account)?; + let page = s + .twitter + .search_mentions(creds, &q.account, q.since_id.as_deref()) + .await + .map_err(|e| { + warn!(account = %q.account, error = %format!("{e:#}"), "mention search failed"); + (StatusCode::BAD_GATEWAY, format!("{e:#}")) + })?; + Ok(Json(MentionsResp { + account: q.account, + newest_id: page.newest_id, + mentions: page + .mentions + .into_iter() + .map(|m| MentionDto { + tweet_id: m.tweet_id, + author_id: m.author_id, + author_handle: m.author_handle, + text: m.text, + created_at: m.created_at, + metrics: m.metrics.into(), + }) + .collect(), + })) +} + +#[derive(Deserialize)] +pub struct TweetMetricsQuery { + /// Account name whose credentials sign the lookup. + pub account: String, + /// Comma-separated tweet ids, at most 100. + pub ids: String, +} + +#[derive(Serialize)] +pub struct TweetMetricsDto { + pub tweet_id: String, + #[serde(flatten)] + pub metrics: MetricsDto, +} + +#[derive(Serialize)] +pub struct TweetMetricsResp { + /// Deleted/protected tweets are absent. + pub metrics: Vec, +} + +/// `GET /tweets/metrics?account=&ids=` — current engagement +/// counters for up to 100 tweets. +pub async fn tweets_metrics( + State(s): State>, + Query(q): Query, +) -> Result, ApiError> { + let creds = account_creds(&s, &q.account)?; + let ids: Vec = q + .ids + .split(',') + .map(str::trim) + .filter(|s| !s.is_empty()) + .map(str::to_string) + .collect(); + if ids.is_empty() || ids.len() > 100 { + return Err(( + StatusCode::BAD_REQUEST, + format!("ids must be 1..=100 comma-separated tweet ids, got {}", ids.len()), + )); + } + + let metrics = s.twitter.tweets_metrics(creds, &ids).await.map_err(|e| { + warn!(account = %q.account, error = %format!("{e:#}"), "metrics lookup failed"); + (StatusCode::BAD_GATEWAY, format!("{e:#}")) + })?; + Ok(Json(TweetMetricsResp { + metrics: metrics + .into_iter() + .map(|t| TweetMetricsDto { + tweet_id: t.tweet_id, + metrics: t.metrics.into(), + }) + .collect(), + })) +} diff --git a/rust-backend/services/twitter-service/src/lib.rs b/rust-backend/services/twitter-service/src/lib.rs index 7880226f..bdf006b5 100644 --- a/rust-backend/services/twitter-service/src/lib.rs +++ b/rust-backend/services/twitter-service/src/lib.rs @@ -12,6 +12,10 @@ //! - `GET /health` //! - `GET /accounts` — the configured account names. //! - `POST /tweets` — `{account, text}` → post a tweet from that account. +//! - `GET /mentions?account=…[&since_id=…]` — recent tweets mentioning +//! `@account` with engagement counters (consumed by engagement-service). +//! - `GET /tweets/metrics?account=…&ids=…` — refresh counters for up to +//! 100 known tweets. pub mod config; pub mod handlers; diff --git a/rust-backend/services/twitter-service/src/router.rs b/rust-backend/services/twitter-service/src/router.rs index 00f097ab..0de91491 100644 --- a/rust-backend/services/twitter-service/src/router.rs +++ b/rust-backend/services/twitter-service/src/router.rs @@ -16,6 +16,8 @@ pub async fn serve(addr: SocketAddr, state: Arc) -> Result<()> { .route("/health", get(handlers::health)) .route("/accounts", get(handlers::accounts)) .route("/tweets", post(handlers::post_tweet)) + .route("/mentions", get(handlers::mentions)) + .route("/tweets/metrics", get(handlers::tweets_metrics)) .with_state(state) .merge(observability::middleware::metrics_route()) .layer(axum::middleware::from_fn( diff --git a/rust-backend/services/twitter-service/src/twitter.rs b/rust-backend/services/twitter-service/src/twitter.rs index 84facb21..cb2304cd 100644 --- a/rust-backend/services/twitter-service/src/twitter.rs +++ b/rust-backend/services/twitter-service/src/twitter.rs @@ -1,9 +1,10 @@ -//! Thin Twitter API v2 client: create-tweet, signed per-account. +//! Thin Twitter API v2 client: create-tweet, recent-mention search and +//! tweet-metrics lookup, signed per-account. use std::collections::BTreeMap; use std::time::Duration; -use anyhow::{anyhow, Context, Result}; +use anyhow::{anyhow, ensure, Context, Result}; use serde::Deserialize; use crate::oauth1; @@ -28,6 +29,92 @@ struct CreateTweetResponse { data: PostedTweet, } +/// Public engagement counters (`tweet.fields=public_metrics`). +#[derive(Debug, Clone, Copy, Default, Deserialize)] +pub struct PublicMetrics { + #[serde(default)] + pub like_count: i64, + #[serde(default)] + pub retweet_count: i64, + #[serde(default)] + pub reply_count: i64, + #[serde(default)] + pub quote_count: i64, +} + +/// One tweet mentioning the account, author resolved from the response's +/// `includes.users`. +#[derive(Debug)] +pub struct Mention { + pub tweet_id: String, + pub author_id: String, + /// Author's @handle, without the `@`. Empty if the expansion was missing. + pub author_handle: String, + pub text: String, + /// RFC 3339, as Twitter returns it. + pub created_at: String, + pub metrics: PublicMetrics, +} + +/// One page of recent-search results (Twitter caps recent search at the +/// last 7 days; `newest_id` is the next poll's `since_id`). +#[derive(Debug)] +pub struct MentionsPage { + pub mentions: Vec, + pub newest_id: Option, +} + +/// Refreshed counters for one tweet (`GET /2/tweets?ids=…`). +#[derive(Debug)] +pub struct TweetMetrics { + pub tweet_id: String, + pub metrics: PublicMetrics, +} + +#[derive(Deserialize)] +struct SearchTweet { + id: String, + text: String, + #[serde(default)] + author_id: String, + #[serde(default)] + created_at: String, + #[serde(default)] + public_metrics: PublicMetrics, +} + +#[derive(Deserialize)] +struct IncludedUser { + id: String, + username: String, +} + +#[derive(Deserialize, Default)] +struct SearchIncludes { + #[serde(default)] + users: Vec, +} + +#[derive(Deserialize)] +struct SearchMeta { + newest_id: Option, +} + +#[derive(Deserialize)] +struct SearchResponse { + #[serde(default)] + data: Vec, + #[serde(default)] + includes: Option, + meta: Option, +} + +#[derive(Deserialize)] +struct LookupResponse { + #[serde(default)] + data: Vec, +} + impl TwitterClient { pub fn new(api_base: &str) -> Result { let http = reqwest::Client::builder() @@ -67,4 +154,122 @@ impl TwitterClient { serde_json::from_str(&body).with_context(|| format!("parsing response: {body}"))?; Ok(parsed.data) } + + /// `GET /2/tweets/search/recent` — original (non-retweet) tweets from + /// the last 7 days mentioning `@handle`, excluding the account's own + /// tweets. The account name in secrets doubles as the handle. + pub async fn search_mentions( + &self, + creds: &TwitterAccount, + handle: &str, + since_id: Option<&str>, + ) -> Result { + let url = format!("{}/2/tweets/search/recent", self.api_base); + let mut params = BTreeMap::new(); + params.insert( + "query".to_string(), + format!("@{handle} -is:retweet -from:{handle}"), + ); + params.insert("max_results".to_string(), "100".to_string()); + params.insert( + "tweet.fields".to_string(), + "public_metrics,created_at,author_id".to_string(), + ); + params.insert("expansions".to_string(), "author_id".to_string()); + params.insert("user.fields".to_string(), "username".to_string()); + if let Some(id) = since_id { + params.insert("since_id".to_string(), id.to_string()); + } + + let body = self + .signed_get(creds, "GET /2/tweets/search/recent", &url, ¶ms) + .await?; + let parsed: SearchResponse = + serde_json::from_str(&body).with_context(|| format!("parsing response: {body}"))?; + + let users: BTreeMap = parsed + .includes + .unwrap_or_default() + .users + .into_iter() + .map(|u| (u.id, u.username)) + .collect(); + let mentions = parsed + .data + .into_iter() + .map(|t| Mention { + author_handle: users.get(&t.author_id).cloned().unwrap_or_default(), + tweet_id: t.id, + author_id: t.author_id, + text: t.text, + created_at: t.created_at, + metrics: t.public_metrics, + }) + .collect(); + Ok(MentionsPage { + mentions, + newest_id: parsed.meta.and_then(|m| m.newest_id), + }) + } + + /// `GET /2/tweets?ids=…` — current engagement counters for up to 100 + /// tweets. Deleted/protected tweets come back under `errors` and are + /// silently absent from the result. + pub async fn tweets_metrics( + &self, + creds: &TwitterAccount, + ids: &[String], + ) -> Result> { + ensure!( + !ids.is_empty() && ids.len() <= 100, + "ids must be 1..=100 per lookup, got {}", + ids.len() + ); + let url = format!("{}/2/tweets", self.api_base); + let mut params = BTreeMap::new(); + params.insert("ids".to_string(), ids.join(",")); + params.insert("tweet.fields".to_string(), "public_metrics".to_string()); + + let body = self.signed_get(creds, "GET /2/tweets", &url, ¶ms).await?; + let parsed: LookupResponse = + serde_json::from_str(&body).with_context(|| format!("parsing response: {body}"))?; + Ok(parsed + .data + .into_iter() + .map(|t| TweetMetrics { + tweet_id: t.id, + metrics: t.public_metrics, + }) + .collect()) + } + + /// Signed GET. Query params are part of the OAuth 1.0a signature base + /// string, so the same map feeds both the header and the URL. + async fn signed_get( + &self, + creds: &TwitterAccount, + op: &'static str, + url: &str, + params: &BTreeMap, + ) -> Result { + let auth = oauth1::authorization_header(creds, "GET", url, params); + let send = |headers| { + self.http + .get(url) + .headers(headers) + .header(reqwest::header::AUTHORIZATION, auth.clone()) + .query(params) + .send() + }; + let resp = observability::client::instrumented("twitter", op, send) + .await + .with_context(|| format!("sending {op} request"))?; + + let status = resp.status(); + let body = resp.text().await.unwrap_or_default(); + if !status.is_success() { + return Err(anyhow!("twitter api {status}: {body}")); + } + Ok(body) + } } From 4fed3ea4621b9444b2b29c860e0ab447668017f6 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 15 Jul 2026 07:24:59 +0000 Subject: [PATCH 2/2] [SO-279] Log PR #281 in PRs.md Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01CacZTy2eX51YUrMNcnpKm7 --- .claude/PRs.md | 1 + 1 file changed, 1 insertion(+) diff --git a/.claude/PRs.md b/.claude/PRs.md index 9d05dede..d7490664 100644 --- a/.claude/PRs.md +++ b/.claude/PRs.md @@ -48,3 +48,4 @@ PRs raised through Claude Code. | [#265](https://github.com/ewitulsk/SuiOptions/pull/265) | [SO-266](https://suioptions.atlassian.net/browse/SO-266) | SO-19 Frontend | Gate Exercise on Expired Options + Fix Off-Screen Popup Positioning | | [#274](https://github.com/ewitulsk/SuiOptions/pull/274) | [SO-272](https://suioptions.atlassian.net/browse/SO-272) | SO-19 Frontend | Fix Put Earn-Page Writer UI (USDC Collateral, Mirrored Outcomes, Dual-Denom Input) | | [#275](https://github.com/ewitulsk/SuiOptions/pull/275) | [SO-273](https://suioptions.atlassian.net/browse/SO-273) | — Gas Station | Sponsor Cash-Secured Put PTBs in Gas-Station Templates | +| [#281](https://github.com/ewitulsk/SuiOptions/pull/281) | [SO-279](https://suioptions.atlassian.net/browse/SO-279) | — Socials | Engagement Tracking, Airdrop Points & Leaderboard MVP (Twitter Mentions + Discord Bot) |