feat(fanout): migrate SlackNotifyFn to FanOutConsumer subscriber (#64)

cdk/src/constructs/fanout-consumer.ts

@@ -20,6 +20,7 @@
 import * as path from 'path';
 import { Duration } from 'aws-cdk-lib';
 import * as dynamodb from 'aws-cdk-lib/aws-dynamodb';
+import * as iam from 'aws-cdk-lib/aws-iam';
 import { StartingPosition, Architecture, Runtime } from 'aws-cdk-lib/aws-lambda';
 import { DynamoEventSource, SqsDlq } from 'aws-cdk-lib/aws-lambda-event-sources';
 import * as lambda from 'aws-cdk-lib/aws-lambda-nodejs';
@@ -62,6 +63,18 @@ export interface FanOutConsumerProps {
    */
   readonly githubTokenSecret?: sm.ISecret;
 
+  /**
+   * Secrets Manager ARN-prefix pattern for per-workspace Slack bot
+   * tokens. Required ONLY when the platform deploys SlackIntegration —
+   * the Slack dispatcher reads bot tokens at this scope. Matches the
+   * other "guarded by prop" grants (taskTable, repoTable,
+   * githubTokenSecret): a deployment without Slack onboarding gets no
+   * dangling IAM permission to ``bgagent/slack/*``. Typically passed
+   * as ``Stack.of(this).formatArn({ ..., resourceName:
+   * 'bgagent/slack/*' })``. Found in PR #79 review (#2 CRITICAL).
+   */
+  readonly slackSecretArnPattern?: string;
+
   /**
    * Maximum batch size delivered to the Lambda per invocation.
    *
@@ -134,6 +147,20 @@ export class FanOutConsumer extends Construct {
       this.fn.addEnvironment('GITHUB_TOKEN_SECRET_ARN', props.githubTokenSecret.secretArn);
     }
 
+    // Slack dispatcher reads per-workspace bot tokens from Secrets
+    // Manager (``bgagent/slack/<team_id>``). Scope the grant to the
+    // caller-provided prefix so the fan-out Lambda cannot read
+    // unrelated platform secrets — matches the policy the old
+    // standalone ``SlackNotifyFn`` held before issue #64. Guarded on
+    // ``slackSecretArnPattern`` so deployments without Slack
+    // onboarding don't get a dangling IAM grant (PR #79 review #2).
+    if (props.slackSecretArnPattern) {
+      this.fn.addToRolePolicy(new iam.PolicyStatement({
+        actions: ['secretsmanager:GetSecretValue'],
+        resources: [props.slackSecretArnPattern],
+      }));
+    }
+
     this.fn.addEventSource(new DynamoEventSource(props.taskEventsTable, {
       startingPosition: StartingPosition.LATEST,
       batchSize: props.batchSize ?? 100,

cdk/src/constructs/slack-integration.ts

@@ -23,8 +23,7 @@ import * as apigw from 'aws-cdk-lib/aws-apigateway';
 import * as cognito from 'aws-cdk-lib/aws-cognito';
 import * as dynamodb from 'aws-cdk-lib/aws-dynamodb';
 import * as iam from 'aws-cdk-lib/aws-iam';
-import { Runtime, Architecture, StartingPosition, FilterCriteria, FilterRule } from 'aws-cdk-lib/aws-lambda';
-import * as lambdaEventSources from 'aws-cdk-lib/aws-lambda-event-sources';
+import { Runtime, Architecture } from 'aws-cdk-lib/aws-lambda';
 import * as lambda from 'aws-cdk-lib/aws-lambda-nodejs';
 import * as secretsmanager from 'aws-cdk-lib/aws-secretsmanager';
 import { NagSuppressions } from 'cdk-nag';
@@ -73,9 +72,14 @@ export interface SlackIntegrationProps {
  * Creates:
  * - SlackInstallationTable (per-workspace installation records)
  * - SlackUserMappingTable (Slack user → platform user mappings)
- * - Lambda handlers for OAuth, slash commands, events, notifications, and account linking
+ * - Lambda handlers for OAuth, slash commands, events, and account linking
  * - API Gateway routes under /slack/*
- * - DynamoDB Streams event source for outbound notifications
+ *
+ * Outbound Slack delivery (task lifecycle notifications) runs through
+ * ``FanOutConsumer`` as a per-channel dispatcher. Before issue #64 this
+ * construct also owned a ``SlackNotifyFn`` DynamoDB Streams consumer on
+ * ``TaskEventsTable``; that consumer was removed to keep the stream at
+ * the DynamoDB-documented one-reader-per-shard limit.
  */
 export class SlackIntegration extends Construct {
   /** The Slack installation table. */
@@ -330,34 +334,8 @@ export class SlackIntegration extends Construct {
     });
     this.userMappingTable.grantReadWriteData(slackLinkFn);
 
-    // --- Outbound Notification Handler (DynamoDB Streams trigger) ---
-    const slackNotifyFn = new lambda.NodejsFunction(this, 'SlackNotifyFn', {
-      entry: path.join(handlersDir, 'slack-notify.ts'),
-      handler: 'handler',
-      runtime: Runtime.NODEJS_24_X,
-      architecture: Architecture.ARM_64,
-      timeout: Duration.seconds(30),
-      environment: {
-        TASK_TABLE_NAME: props.taskTable.tableName,
-      },
-      bundling: commonBundling,
-    });
-    props.taskTable.grantReadWriteData(slackNotifyFn);
-    slackNotifyFn.addToRolePolicy(readSlackSecretsPolicy);
-
-    // DynamoDB Streams event source with filtering
-    slackNotifyFn.addEventSource(new lambdaEventSources.DynamoEventSource(props.taskEventsTable, {
-      startingPosition: StartingPosition.LATEST,
-      batchSize: 10,
-      maxBatchingWindow: Duration.seconds(0),
-      retryAttempts: 3,
-      bisectBatchOnError: true,
-      filters: [
-        FilterCriteria.filter({
-          eventName: FilterRule.isEqual('INSERT'),
-        }),
-      ],
-    }));
+    // Outbound Slack delivery runs through FanOutConsumer — see the
+    // construct doc above for the reader-count rationale (issue #64).
 
     // ═══════════════════════════════════════════════════════════════════════════
     // API Gateway Routes
@@ -436,7 +414,7 @@ export class SlackIntegration extends Construct {
     }
 
     // Standard Lambda suppressions
-    const allFunctions = [oauthCallbackFn, slackEventsFn, slackCommandsFn, commandProcessorFn, slackLinkFn, slackNotifyFn, slackInteractionsFn];
+    const allFunctions = [oauthCallbackFn, slackEventsFn, slackCommandsFn, commandProcessorFn, slackLinkFn, slackInteractionsFn];
     for (const fn of allFunctions) {
       NagSuppressions.addResourceSuppressions(fn, [
         {