Dev

.env.example

@@ -3,6 +3,7 @@
 
 # === REQUIRED CONFIGURATION ===
 ANTICAPTURE_GRAPHQL_ENDPOINT=https://api-gateway-production-0879.up.railway.app/graphql
+BLOCKFUL_API_TOKEN=
 
 # === NOTIFICATION PLATFORMS ===
 TELEGRAM_BOT_TOKEN=

AGENTS.md

@@ -181,39 +181,68 @@ ALTER TABLE votes_onchain ENABLE TRIGGER ALL;
 ```
 
 ### Voting Power Change Insert (Delegation Received)
+
+**Important:** The underlying tables are in a UUID schema (e.g. `ecbe454c-b8fe-4659-8864-4cc68148cfde`), not in `anticapture` (which has views). Find the correct schema via: `SELECT pg_get_viewdef('anticapture.voting_power_history', true);`
+
+**Critical:** Use a **fixed transaction hash** (not `extract(epoch from now())`) so both tables share the exact same value. The API joins `voting_power_history` with `delegations` on `transaction_hash`, and requires `delegation.log_index < voting_power_history.log_index`. If the join fails, `changeType` becomes `'other'`, which bypasses threshold filtering entirely.
+
 ```sql
-SET search_path TO "<schema_uuid>";
+SET search_path TO "<underlying_uuid_schema>";
 ALTER TABLE voting_power_history DISABLE TRIGGER ALL;
 ALTER TABLE delegations DISABLE TRIGGER ALL;
 
-INSERT INTO voting_power_history (transaction_hash, dao_id, account_id, voting_power, delta, delta_mod, timestamp, log_index)
+-- Insert delegations FIRST with log_index = 0
+INSERT INTO delegations (transaction_hash, dao_id, delegate_account_id, delegator_account_id, delegated_value, previous_delegate, timestamp, log_index)
 VALUES (
-  '0xmock_vp_' || extract(epoch from now())::bigint,
+  '0xmock_vp_test',
   'ENS',
   '<user_wallet_address>',
-  5000000000000000000,
-  1000000000000000000,
+  '0x1111111111111111111111111111111111111111',
   1000000000000000000,
+  '0x0000000000000000000000000000000000000000',
   extract(epoch from now())::bigint,
-  1
+  0  -- must be < voting_power_history.log_index
 );
 
-INSERT INTO delegations (transaction_hash, dao_id, delegate_account_id, delegator_account_id, delegated_value, previous_delegate, timestamp, log_index)
+-- Then insert voting_power_history with log_index = 1
+INSERT INTO voting_power_history (transaction_hash, dao_id, account_id, voting_power, delta, delta_mod, timestamp, log_index)
 VALUES (
-  '0xmock_vp_' || extract(epoch from now())::bigint,
+  '0xmock_vp_test',
   'ENS',
   '<user_wallet_address>',
-  '0x1111111111111111111111111111111111111111',
+  5000000000000000000,
+  1000000000000000000,
   1000000000000000000,
-  '0x0000000000000000000000000000000000000000',
   extract(epoch from now())::bigint,
-  1
+  1  -- must be > delegation.log_index
 );
 
 ALTER TABLE voting_power_history ENABLE TRIGGER ALL;
 ALTER TABLE delegations ENABLE TRIGGER ALL;
 ```
 
+### New Offchain Proposal Insert (Snapshot)
+```sql
+-- No triggers on snapshot.proposals, safe to insert directly
+INSERT INTO snapshot.proposals (id, space_id, author, title, body, discussion, type, start, "end", state, created, updated, link, flagged)
+VALUES (
+  '0xmock_offchain_' || extract(epoch from now())::bigint,
+  'ens.eth',
+  '0x1111111111111111111111111111111111111111',
+  '[MOCK] Test Offchain Proposal for Notification System',
+  'This is a mock offchain proposal inserted for testing the notification pipeline.',
+  '<discussion_url_from_existing_proposal>',  -- e.g. https://discuss.ens.domains/t/...
+  'single-choice',
+  extract(epoch from now())::integer,
+  (extract(epoch from now()) + 604800)::integer,  -- ends in 7 days
+  'active',
+  extract(epoch from now())::integer,
+  extract(epoch from now())::integer,
+  '<link_from_existing_proposal>',  -- copy from: SELECT link FROM snapshot.proposals ORDER BY created DESC LIMIT 5
+  false
+);
+```
+
 ### Cleanup Mock Data
 ```sql
 SET search_path TO "<schema_uuid>";
@@ -224,6 +253,7 @@ ALTER TABLE delegations DISABLE TRIGGER ALL;
 DELETE FROM votes_onchain WHERE tx_hash LIKE '0xmock%';
 DELETE FROM voting_power_history WHERE transaction_hash LIKE '0xmock%';
 DELETE FROM delegations WHERE transaction_hash LIKE '0xmock%';
+DELETE FROM snapshot.proposals WHERE id LIKE '0xmock%';
 
 ALTER TABLE votes_onchain ENABLE TRIGGER ALL;
 ALTER TABLE voting_power_history ENABLE TRIGGER ALL;
@@ -241,5 +271,6 @@ SELECT v.* FROM votes_onchain v LEFT JOIN proposals_onchain p ON v.proposal_id =
 -- List mock records
 SELECT 'VOTE' as type, tx_hash, timestamp FROM votes_onchain WHERE tx_hash LIKE '0xmock%'
 UNION ALL SELECT 'VP', transaction_hash, timestamp FROM voting_power_history WHERE transaction_hash LIKE '0xmock%'
-UNION ALL SELECT 'DELEG', transaction_hash, timestamp FROM delegations WHERE transaction_hash LIKE '0xmock%';
+UNION ALL SELECT 'DELEG', transaction_hash, timestamp FROM delegations WHERE transaction_hash LIKE '0xmock%'
+UNION ALL SELECT 'OFFCHAIN', id, created FROM snapshot.proposals WHERE id LIKE '0xmock%';
 ```

apps/consumers/example.env

@@ -1,4 +1,5 @@
 TELEGRAM_BOT_TOKEN=your_telegram_bot_token
 ANTICAPTURE_GRAPHQL_ENDPOINT=https://api-gateway/graphql
+BLOCKFUL_API_TOKEN=
 SUBSCRIPTION_SERVER_URL=http://localhost:3001
 API_PORT=3000

apps/consumers/src/config/env.ts

@@ -10,9 +10,11 @@ const envSchema = z.object({
   SLACK_SIGNING_SECRET: z.string(),
   TOKEN_ENCRYPTION_KEY: z.string(),
   ANTICAPTURE_GRAPHQL_ENDPOINT: z.string().url("ANTICAPTURE_GRAPHQL_ENDPOINT must be a valid URL"),
+  BLOCKFUL_API_TOKEN: z.string().optional(),
   SUBSCRIPTION_SERVER_URL: z.string(),
   RABBITMQ_URL: z.string().url(),
   PORT: z.coerce.number().positive().optional().default(3002),
+  RPC_URL: z.string().optional(),
 });
 
 export function loadConfig() {
@@ -24,8 +26,10 @@ export function loadConfig() {
     slackSigningSecret: env.SLACK_SIGNING_SECRET,
     tokenEncryptionKey: env.TOKEN_ENCRYPTION_KEY,
     anticaptureGraphqlEndpoint: env.ANTICAPTURE_GRAPHQL_ENDPOINT,
+    blockfulApiToken: env.BLOCKFUL_API_TOKEN,
     subscriptionServerUrl: env.SUBSCRIPTION_SERVER_URL,
     rabbitmqUrl: env.RABBITMQ_URL,
     port: env.PORT,
+    rpcUrl: env.RPC_URL
   } as const;
 } 

apps/consumers/src/index.ts

@@ -19,7 +19,7 @@ import { SlackClient } from './clients/slack.client';
 const config = loadConfig();
 
 // Create ENS resolver
-const ensResolver = new EnsResolverService();
+const ensResolver = new EnsResolverService(config.rpcUrl);
 
 // Create Telegram client for production
 const telegramClient = new TelegramClient(config.telegramBotToken);
@@ -35,7 +35,14 @@ const slackClient = new SlackClient(
 // Create and start the application
 const app = new App(
   config.subscriptionServerUrl,
-  axios.create({ baseURL: config.anticaptureGraphqlEndpoint }),
+  axios.create({
+    baseURL: config.anticaptureGraphqlEndpoint,
+    headers: {
+      ...(config.blockfulApiToken && {
+        Authorization: `Bearer ${config.blockfulApiToken}`,
+      }),
+    },
+  }),
   config.rabbitmqUrl,
   ensResolver,
   telegramClient,

apps/consumers/src/interfaces/bot.interface.ts

@@ -10,6 +10,7 @@ export interface ContextWithSession extends Context {
         walletAction?: 'add' | 'remove';
         walletsToRemove?: Set<string>;
         awaitingWalletInput?: boolean;
+        fromStart?: boolean;
     };
 }
 

apps/consumers/src/interfaces/slack-context.interface.ts

@@ -33,6 +33,7 @@ export interface SlackBodyWithIds {
       };
     };
   } | string; // Can be string for DialogSubmitAction or object for BlockAction
+  actions?: Array<{ action_id?: string; value?: string; type?: string }>;
 }
 
 /**
@@ -56,6 +57,7 @@ export interface SlackSession {
     type: 'wallet' | 'dao';
     action: 'add' | 'remove';
   };
+  fromStart?: boolean;
 }
 
 /**

apps/consumers/src/services/bot/slack-bot.service.ts

@@ -8,7 +8,7 @@
 import { SlackClientInterface } from '../../interfaces/slack-client.interface';
 import { NotificationPayload } from '../../interfaces/notification.interface';
 import { BotServiceInterface } from '../../interfaces/bot-service.interface';
-import { slackMessages, convertMarkdownToSlack } from '@notification-system/messages';
+import { slackMessages, convertMarkdownToSlack, appendUtmParams } from '@notification-system/messages';
 import { EnsResolverService } from '../ens-resolver.service';
 import { SlackDAOService } from '../dao/slack-dao.service';
 import { SlackWalletService } from '../wallet/slack-wallet.service';
@@ -48,6 +48,11 @@ export class SlackBotService implements BotServiceInterface {
       // Welcome message actions
       handlers.action('welcome_select_daos', async (ctx) => {
         if (this.daoService) {
+          const channelId = ctx.body.channel?.id;
+          const workspaceId = ctx.body.team?.id || ctx.body.user?.team_id;
+          const fullUserId = `${workspaceId}:${channelId}`;
+          const hasDaos = await this.daoService.hasSubscriptions(fullUserId);
+          ctx.session.fromStart = !hasDaos;
           await this.daoService.initialize(ctx);
         }
       });
@@ -74,13 +79,20 @@ export class SlackBotService implements BotServiceInterface {
       handlers.action('dao_confirm_subscribe', async (ctx) => {
         if (this.daoService) {
           await this.daoService.confirm(ctx);
+
+          // If from onboarding flow, trigger wallet setup
+          if (ctx.session.fromStart && this.walletService) {
+            await this.walletService.showOnboardingWallet(ctx);
+            ctx.session.fromStart = false;
+          }
         }
       });
 
       handlers.action('dao_checkboxes', async (ctx) => {
         await ctx.ack();
       });
 
+
       handlers.action('wallet_checkboxes', async (ctx) => {
         await ctx.ack();
       });
@@ -94,6 +106,8 @@ export class SlackBotService implements BotServiceInterface {
 
       handlers.action('wallet_add', async (ctx) => {
         if (this.walletService) {
+          const firstAction = ctx.body.actions?.[0];
+          ctx.session.fromStart = firstAction?.value === 'onboarding';
           await this.walletService.startAddWallet(ctx);
         }
       });
@@ -212,16 +226,25 @@ export class SlackBotService implements BotServiceInterface {
       throw new Error('Slack notification requires workspace OAuth token. No bot_token provided in notification payload.');
     }
 
+    // Append UTM tracking params to button URLs
+    const triggerType = payload.metadata?.triggerType;
+    const buttons = payload.metadata?.buttons?.map(btn => ({
+      text: btn.text,
+      url: triggerType
+        ? appendUtmParams(btn.url, { source: 'notification', medium: 'slack', campaign: triggerType })
+        : btn.url
+    }));
+
     // Build message options with buttons if provided
-    const messageOptions = payload.metadata?.buttons ? {
+    const messageOptions = buttons ? {
       blocks: [
         {
           type: 'section' as const,
           text: { type: 'mrkdwn' as const, text: processedMessage }
         },
         {
           type: 'actions' as const,
-          elements: payload.metadata.buttons.map(btn => ({
+          elements: buttons.map(btn => ({
             type: 'button' as const,
             text: { type: 'plain_text' as const, text: btn.text },
             url: btn.url