Developer Handover Pack

Developer Handover Pack

Data Quality Monitoring

Connects to PostgreSQL and Airtable on a daily schedule, or immediately on receipt of a data import event. Runs the full configured quality ruleset against the latest records in all watched tables and fields. Evaluates each record for null values, format mismatches, out-of-range figures, and duplicate keys. Assigns a severity score (Critical, High, Moderate, Low) to every issue found based on the pre-agreed scoring thresholds. Produces a structured issue list and passes it downstream to the Issue Routing and Notification Agent. Replaces the four most time-consuming manual steps in the current process and eliminates the scheduling dependency entirely.

// Input
trigger_type: 'schedule' | 'import_event'
import_event.source_table: string          // e.g. 'customers', 'orders'
import_event.row_count: integer
import_event.timestamp: ISO8601
ruleset_config: GoogleSheets.tab('Rules')  // field, rule_type, threshold, severity_weight

// Processing
PostgreSQL.query(watched_tables, watched_fields) -> raw_records[]
Airtable.select(base='IssueLog', view='ActiveSources') -> source_metadata[]
rules_engine.evaluate(raw_records, ruleset_config) -> flagged_records[]
severity_scorer.score(flagged_records, severity_weights) -> scored_issues[]

// Output
issue_list[]: {
  source_table: string,
  field_name: string,
  error_type: 'null' | 'format' | 'duplicate' | 'out_of_range' | 'referential',
  affected_row_count: integer,
  severity: 'Critical' | 'High' | 'Moderate' | 'Low',
  severity_score: float,
  sample_record_id: string,
  detected_at: ISO8601
}

Receives the completed issue list from the Data Check Agent at the end of each check run. Creates Jira tickets for all Critical and High severity issues, pre-filled with the source table, field name, error type, affected row count, severity score, and a direct link to the Airtable log row. Writes every issue, regardless of severity, to the Airtable Issues Log as a new or updated row. Posts a formatted summary message to the designated IT Slack channel listing severity counts, the top three Critical issues by score, and links to all Jira tickets created in the run. Replaces the three manual communication and logging steps that previously consumed 45 minutes per run.

// Input
issue_list[]: {                            // received from Data Check Agent output
  source_table, field_name, error_type,
  affected_row_count, severity, severity_score,
  sample_record_id, detected_at
}
jira_project_key: string                  // e.g. 'DQ' — must be confirmed pre-build
slack_channel_id: string                  // e.g. 'C08XXXXX' — IT channel webhook
airtable_base_id: string                  // Issues Log base ID

// Processing
filter(issue_list, severity IN ['Critical','High']) -> ticket_candidates[]
Jira.createIssue(ticket_candidates) -> jira_tickets[]: { ticket_id, ticket_url }
Airtable.upsert(issue_list + jira_ticket_urls) -> airtable_rows[]
Slack.postMessage(channel_id, format_summary(issue_list, jira_tickets)) -> message_ts

// Output
jira_tickets[]: {
  ticket_id: string,
  ticket_url: string,
  linked_issue: { source_table, field_name, severity }
}
airtable_rows[]: {
  airtable_record_id: string,
  status: 'Open' | 'Updated',
  jira_ticket_url: string | null
}
slack_message_ts: string                  // for thread replies if needed

Fires on a weekly schedule after the final daily check of the week completes. Queries the Airtable Issues Log to aggregate the past seven days of data, computing total issues detected, issues by severity, issues resolved within SLA, and open carry-over count. Pushes all metrics to the Datadog dashboard via the Datadog API, updating the agreed metric series so the IT Manager has a live trend view without any manual compilation. Replaces the 20-minute weekly summary report that the analyst previously assembled by hand from spreadsheet totals.

// Input
airtable_base_id: string                  // Issues Log base
date_range: { start: ISO8601, end: ISO8601 } // past 7 days
datadog_api_key: string                   // from credential store
datadog_app_key: string                   // from credential store
metric_namespace: string                  // e.g. 'dq.weekly'

// Processing
Airtable.select(base=airtable_base_id, filter='detected_at >= start AND detected_at <= end')
  -> weekly_issues[]: { severity, status, resolution_time_hours, source_table }
aggregate(weekly_issues) -> metrics: {
  total_issues_detected: integer,
  critical_count: integer,
  high_count: integer,
  moderate_count: integer,
  low_count: integer,
  resolved_count: integer,
  open_count: integer,
  avg_resolution_hours: float,
  issues_by_source: { [source_table]: integer }
}
Datadog.submitMetrics(metric_namespace, metrics, timestamp=week_end)

// Output
datadog_series_ids[]: string[]            // confirmation of accepted metric series
summary_log_row: {                        // written back to Airtable for audit
  week_ending: ISO8601,
  total_detected: integer,
  total_resolved: integer,
  pushed_to_datadog: boolean
}

// ============================================================
// TRIGGER LAYER
// ============================================================
TRIGGER: cron('0 6 * * *')                // daily at 06:00 UTC
      OR webhook.receive(event='import_complete', source_table, row_count, timestamp)

trigger_context = {
  trigger_type: 'schedule' | 'import_event',
  source_table: string | null,
  row_count: integer | null,
  fired_at: ISO8601
}

// ============================================================
// AGENT 1: DATA CHECK AGENT
// ============================================================

// Step 1: Load ruleset from Google Sheets
GoogleSheets.read(sheet='Rules', tab='ActiveRules')
  -> ruleset[]: { field_name, rule_type, threshold, severity_weight, source_table }

// Step 2: Query data sources
PostgreSQL.query(
  tables: ruleset[].source_table (distinct),
  fields: ruleset[].field_name (per table),
  limit: configurable (default 50000 rows per table)
) -> raw_records[]: { record_id, source_table, field_name, field_value }

Airtable.select(
  base: airtable_base_id,
  view: 'ActiveSources'
) -> source_metadata[]: { source_name, last_checked_at, expected_row_floor }

// Step 3: Run rules engine
rules_engine.evaluate(raw_records, ruleset)
  -> flagged_records[]: {
       record_id, source_table, field_name,
       rule_violated: rule_type,
       observed_value: string,
       expected_constraint: string
     }

// Step 4: Score severity
severity_scorer.score(flagged_records, severity_weights)
  -> scored_issues[]: {
       source_table, field_name, error_type,
       affected_row_count, severity, severity_score,
       sample_record_id, detected_at: ISO8601
     }

// -- HANDOFF: Data Check Agent -> Issue Routing Agent --
// Payload: scored_issues[]
// Condition: always fires on run completion (even if scored_issues is empty)

// ============================================================
// AGENT 2: ISSUE ROUTING AND NOTIFICATION AGENT
// ============================================================

// Step 5: Write all issues to Airtable Issues Log
Airtable.upsert(
  base: airtable_base_id,
  table: 'IssuesLog',
  match_fields: ['source_table','field_name','error_type','detected_date'],
  payload: scored_issues[]
) -> airtable_rows[]: { airtable_record_id, status: 'Open'|'Updated', jira_ticket_url }

// Step 6: Create Jira tickets for Critical and High issues
ticket_candidates = filter(scored_issues, severity IN ['Critical','High'])
Jira.createIssue(
  project_key: jira_project_key,
  issue_type: 'Bug',
  summary: '[DQ] {source_table}.{field_name} — {error_type} ({affected_row_count} rows)',
  description: { field_name, error_type, affected_row_count, severity_score,
                  airtable_row_url, sample_record_id, detected_at },
  priority: severity -> Jira priority map { Critical:'Highest', High:'High' },
  labels: ['data-quality', source_table]
) -> jira_tickets[]: { ticket_id, ticket_url, linked_issue }

// Step 7: Update Airtable rows with Jira ticket URLs
Airtable.update(
  records: airtable_rows (Critical + High only),
  fields: { jira_ticket_url, jira_status: 'Created' }
)

// Step 8: Post Slack summary to IT channel
Slack.postMessage(
  channel: slack_channel_id,
  blocks: [
    header: 'Data Quality Check — {fired_at date}',
    counts: { critical, high, moderate, low },
    top_critical: scored_issues.filter(Critical).sortDesc(severity_score).slice(0,3)
               .map({ field_name, source_table, affected_row_count, ticket_url }),
    footer: 'Full log: {airtable_log_url}'
  ]
) -> slack_message_ts: string

// -- HANDOFF: Issue Routing Agent ends daily cycle --
// Retained step: IT Analyst reviews Jira tickets, applies manual fixes to PostgreSQL

// ============================================================
// AGENT 3: REPORTING AND METRICS AGENT (weekly trigger only)
// ============================================================

TRIGGER: cron('0 7 * * 5')               // Friday at 07:00 UTC (after daily run)

// Step 9: Aggregate weekly data from Airtable
Airtable.select(
  base: airtable_base_id,
  table: 'IssuesLog',
  filter: 'detected_at >= week_start AND detected_at <= week_end'
) -> weekly_issues[]: { severity, status, resolution_time_hours, source_table }

metrics = {
  'dq.weekly.total_detected': count(weekly_issues),
  'dq.weekly.critical_count': count(severity='Critical'),
  'dq.weekly.high_count':     count(severity='High'),
  'dq.weekly.moderate_count': count(severity='Moderate'),
  'dq.weekly.low_count':      count(severity='Low'),
  'dq.weekly.resolved_count': count(status='Resolved'),
  'dq.weekly.open_count':     count(status='Open'),
  'dq.weekly.avg_resolution_hours': avg(resolution_time_hours)
}

// Step 10: Push metrics to Datadog
Datadog.submitMetrics(
  api_key: datadog_api_key,
  series: metrics.map({ metric: key, points: [[week_end_unix, value]], type: 'gauge' })
) -> datadog_series_ids[]

// Step 11: Write audit row to Airtable
Airtable.insert(
  table: 'WeeklyReports',
  row: { week_ending, total_detected, total_resolved, pushed_to_datadog: true }
)