Add payroll notification Lambda for Gusto payroll via Dataddo

Queries Aurora for new payroll runs and contractor payments, sends
formatted Slack message with gross pay breakdown and total bank
withdrawal amount. Runs weekdays at 2pm ET, tracks notified payrolls
in DynamoDB to avoid duplicates.

package.json

@@ -8,6 +8,7 @@
   ],
   "dependencies": {
     "@aws-sdk/client-dynamodb": "^3.600.0",
+    "@aws-sdk/client-rds-data": "^3.1037.0",
     "@aws-sdk/client-s3": "^3.600.0",
     "@aws-sdk/client-ssm": "^3.600.0",
     "@aws-sdk/lib-dynamodb": "^3.600.0",

src/notifyPayroll.js

@@ -0,0 +1,187 @@
+import { RDSDataClient, ExecuteStatementCommand } from "@aws-sdk/client-rds-data";
+import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
+import { DynamoDBDocumentClient, GetCommand, PutCommand } from "@aws-sdk/lib-dynamodb";
+import { SSMClient, GetParameterCommand } from "@aws-sdk/client-ssm";
+
+const rds = new RDSDataClient();
+const ddb = DynamoDBDocumentClient.from(new DynamoDBClient());
+const ssm = new SSMClient();
+
+const CLUSTER_ARN = process.env.AURORA_CLUSTER_ARN;
+const SECRET_ARN = process.env.AURORA_SECRET_ARN;
+const DATABASE = "payroll";
+const TABLE_NAME = process.env.TABLE_NAME;
+const CHANNEL_ID = "C0AV5RBMYKU";
+
+let cachedToken;
+async function getSlackToken() {
+  if (cachedToken) return cachedToken;
+  const { Parameter } = await ssm.send(
+    new GetParameterCommand({ Name: process.env.SLACK_BOT_TOKEN_PARAM, WithDecryption: true })
+  );
+  cachedToken = Parameter.Value;
+  return cachedToken;
+}
+
+async function sql(query) {
+  const { records } = await rds.send(
+    new ExecuteStatementCommand({
+      resourceArn: CLUSTER_ARN,
+      secretArn: SECRET_ARN,
+      database: DATABASE,
+      sql: query,
+    })
+  );
+  return records || [];
+}
+
+function extractValue(field) {
+  if (field.stringValue !== undefined) return field.stringValue;
+  if (field.longValue !== undefined) return field.longValue;
+  if (field.doubleValue !== undefined) return field.doubleValue;
+  if (field.booleanValue !== undefined) return field.booleanValue;
+  if (field.isNull) return null;
+  return null;
+}
+
+const formatCurrency = (v) =>
+  new Intl.NumberFormat("en-US", { style: "currency", currency: "USD" }).format(Number(v || 0));
+
+const formatDate = (ts) => {
+  const d = new Date(ts);
+  return d.toLocaleDateString("en-US", { weekday: "short", month: "short", day: "numeric", year: "numeric" });
+};
+
+async function getNotifiedPayrolls() {
+  const result = await ddb.send(
+    new GetCommand({ TableName: TABLE_NAME, Key: { pk: "PAYROLL_NOTIFIED" } })
+  );
+  return result.Item?.checkDates || [];
+}
+
+async function markNotified(checkDate) {
+  const existing = await getNotifiedPayrolls();
+  await ddb.send(
+    new PutCommand({
+      TableName: TABLE_NAME,
+      Item: { pk: "PAYROLL_NOTIFIED", checkDates: [...existing, checkDate] },
+    })
+  );
+}
+
+async function buildPayrollMessage(row) {
+  const checkDate = extractValue(row[0]);
+  const companyDebit = extractValue(row[1]);
+  const grossPay = extractValue(row[2]);
+  const netPay = extractValue(row[3]);
+  const employerTaxes = extractValue(row[4]);
+  const reimbursements = extractValue(row[5]);
+  const benefits = extractValue(row[6]);
+
+  const contractorRows = await sql(
+    `SELECT COALESCE(SUM(payments_wage_total), 0), COALESCE(SUM(payments_reimbursement), 0) FROM contractor_payments WHERE payments_date = '${checkDate}'`
+  );
+  const contractorWages = contractorRows.length > 0 ? extractValue(contractorRows[0][0]) : 0;
+  const contractorReimbursements = contractorRows.length > 0 ? extractValue(contractorRows[0][1]) : 0;
+  const contractorTotal = contractorWages + contractorReimbursements;
+
+  const totalGross = grossPay + contractorTotal;
+  const cashRequirement = companyDebit + contractorTotal + benefits;
+
+  const blocks = [
+    {
+      type: "header",
+      text: { type: "plain_text", text: `Payroll Processed — ${formatDate(checkDate)}` },
+    },
+    {
+      type: "section",
+      fields: [
+        { type: "mrkdwn", text: `*Check Date*\n${formatDate(checkDate)}` },
+        { type: "mrkdwn", text: `*Total Gross Pay*\n${formatCurrency(totalGross)}` },
+      ],
+    },
+    { type: "divider" },
+    {
+      type: "section",
+      fields: [
+        { type: "mrkdwn", text: `*Employee Gross*\n${formatCurrency(grossPay)}` },
+        { type: "mrkdwn", text: `*Contractor Gross*\n${formatCurrency(contractorTotal)}` },
+      ],
+    },
+    {
+      type: "section",
+      fields: [
+        { type: "mrkdwn", text: `*Employee Net Pay*\n${formatCurrency(netPay)}` },
+        { type: "mrkdwn", text: `*Reimbursements*\n${formatCurrency(reimbursements)}` },
+      ],
+    },
+    {
+      type: "section",
+      fields: [
+        { type: "mrkdwn", text: `*Employer Taxes*\n${formatCurrency(employerTaxes)}` },
+        { type: "mrkdwn", text: `*Benefits (401k)*\n${formatCurrency(benefits)}` },
+      ],
+    },
+    { type: "divider" },
+    {
+      type: "section",
+      text: {
+        type: "mrkdwn",
+        text: `:moneybag: *Total Bank Withdrawal on ${formatDate(checkDate)}: ${formatCurrency(cashRequirement)}*`,
+      },
+    },
+  ];
+
+  return { checkDate, blocks };
+}
+
+export const handler = async () => {
+  const payrollRows = await sql(
+    "SELECT check_date, company_debit, gross_pay, net_pay, employer_taxes, reimbursements, benefits FROM payroll WHERE company_debit > 0 ORDER BY check_date DESC"
+  );
+
+  if (payrollRows.length === 0) {
+    console.log("No processed payrolls found");
+    return { statusCode: 200, body: "No payrolls" };
+  }
+
+  const notified = await getNotifiedPayrolls();
+  const newPayrolls = payrollRows.filter(
+    (row) => !notified.includes(extractValue(row[0]))
+  );
+
+  if (newPayrolls.length === 0) {
+    console.log("All payrolls already notified");
+    return { statusCode: 200, body: "Already notified" };
+  }
+
+  const token = await getSlackToken();
+
+  for (const row of newPayrolls) {
+    const { checkDate, blocks } = await buildPayrollMessage(row);
+
+    const res = await fetch("https://slack.com/api/chat.postMessage", {
+      method: "POST",
+      headers: {
+        Authorization: `Bearer ${token}`,
+        "Content-Type": "application/json",
+      },
+      body: JSON.stringify({
+        channel: CHANNEL_ID,
+        blocks,
+        text: `Payroll processed for ${formatDate(checkDate)}`,
+      }),
+    });
+
+    const data = await res.json();
+    if (!data.ok) {
+      console.error("Slack post failed:", data.error);
+      throw new Error(`Slack API error: ${data.error}`);
+    }
+
+    await markNotified(checkDate);
+    console.log(`Notified for check date: ${checkDate}`);
+  }
+
+  return { statusCode: 200, body: `Notified ${newPayrolls.length} payroll(s)` };
+};

template.yaml

@@ -330,6 +330,51 @@ Resources:
                 - ec2:DeleteNetworkInterface
               Resource: "*"
 
+  NotifyPayrollFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      FunctionName: payments-notifyPayroll
+      Handler: src/notifyPayroll.handler
+      Timeout: 60
+      VpcConfig:
+        SubnetIds:
+          - !Ref PrivateSubnet
+        SecurityGroupIds:
+          - !Ref LambdaSecurityGroup
+      Environment:
+        Variables:
+          AURORA_CLUSTER_ARN: !GetAtt AuroraCluster.DBClusterArn
+          AURORA_SECRET_ARN: !GetAtt AuroraCluster.MasterUserSecret.SecretArn
+          SLACK_BOT_TOKEN_PARAM: /payments-dashboard/slack-bot-token
+      Events:
+        PayrollCheck:
+          Type: Schedule
+          Properties:
+            Schedule: cron(0 18 ? * MON-FRI *)
+            Description: Check for new payroll data at 2pm ET (18:00 UTC)
+            Enabled: true
+      Policies:
+        - DynamoDBCrudPolicy:
+            TableName: !Ref DashboardTable
+        - SSMParameterReadPolicy:
+            ParameterName: payments-dashboard/slack-bot-token
+        - Version: "2012-10-17"
+          Statement:
+            - Effect: Allow
+              Action:
+                - ec2:CreateNetworkInterface
+                - ec2:DescribeNetworkInterfaces
+                - ec2:DeleteNetworkInterface
+              Resource: "*"
+            - Effect: Allow
+              Action:
+                - rds-data:ExecuteStatement
+              Resource: !GetAtt AuroraCluster.DBClusterArn
+            - Effect: Allow
+              Action:
+                - secretsmanager:GetSecretValue
+              Resource: !GetAtt AuroraCluster.MasterUserSecret.SecretArn
+
 Outputs:
   SlackEventUrl:
     Description: URL to set as the Slack app Request URL