diff --git a/infra/deployments/forms/health/grafana.tf b/infra/deployments/forms/health/grafana.tf
new file mode 100644
index 000000000..447c901df
--- /dev/null
+++ b/infra/deployments/forms/health/grafana.tf
@@ -0,0 +1,47 @@
+locals {
+  managed_grafana_count = var.environmental_settings.enable_managed_grafana ? 1 : 0
+}
+
+data "aws_iam_policy_document" "grafana_assume_role" {
+  statement {
+    actions = ["sts:AssumeRole"]
+
+    principals {
+      type        = "Service"
+      identifiers = ["grafana.amazonaws.com"]
+    }
+  }
+}
+
+resource "aws_iam_role" "grafana" {
+  count = local.managed_grafana_count
+
+  name               = "${var.environment_name}-grafana-workspace"
+  assume_role_policy = data.aws_iam_policy_document.grafana_assume_role.json
+}
+
+resource "aws_iam_role_policy_attachment" "grafana" {
+  for_each = var.environmental_settings.enable_managed_grafana ? toset([
+    "arn:aws:iam::aws:policy/service-role/AmazonGrafanaCloudWatchAccess",
+    "arn:aws:iam::aws:policy/service-role/AmazonGrafanaAthenaAccess",
+    "arn:aws:iam::aws:policy/AWSXrayReadOnlyAccess",
+  ]) : toset([])
+
+  role       = aws_iam_role.grafana[0].name
+  policy_arn = each.value
+}
+
+resource "aws_grafana_workspace" "this" {
+  count = local.managed_grafana_count
+
+  name                     = "${var.environment_name}-forms"
+  description              = "GOV.UK Forms ${var.environment_name} observability"
+  account_access_type      = "CURRENT_ACCOUNT"
+  authentication_providers = ["AWS_SSO"]
+  permission_type          = "SERVICE_MANAGED"
+  role_arn                 = aws_iam_role.grafana[0].arn
+  data_sources             = ["ATHENA", "CLOUDWATCH", "XRAY"]
+  grafana_version          = "12.4"
+
+  depends_on = [aws_iam_role_policy_attachment.grafana]
+}
diff --git a/infra/deployments/forms/health/outputs.tf b/infra/deployments/forms/health/outputs.tf
new file mode 100644
index 000000000..669664a64
--- /dev/null
+++ b/infra/deployments/forms/health/outputs.tf
@@ -0,0 +1,3 @@
+output "grafana_workspace_endpoint" {
+  value = one(aws_grafana_workspace.this[*].endpoint)
+}
diff --git a/infra/deployments/forms/health/submission-events.tf b/infra/deployments/forms/health/submission-events.tf
new file mode 100644
index 000000000..e7fca1997
--- /dev/null
+++ b/infra/deployments/forms/health/submission-events.tf
@@ -0,0 +1,5 @@
+module "submission_events" {
+  count    = var.forms_runner_settings.enable_submission_events_analytics ? 1 : 0
+  source   = "../../../modules/submission-events"
+  env_name = var.environment_name
+}
diff --git a/infra/deployments/forms/inputs.tf b/infra/deployments/forms/inputs.tf
index 9dcb14624..658e770fc 100644
--- a/infra/deployments/forms/inputs.tf
+++ b/infra/deployments/forms/inputs.tf
@@ -178,6 +178,7 @@ variable "forms_runner_settings" {
     govuk_one_login_base_url                                        = string
     queue_worker_capacity                                           = string
     disable_builtin_solidqueue_worker                               = bool
+    enable_submission_events_analytics                              = optional(bool, false)
   })
 }
 
@@ -206,6 +207,7 @@ variable "environmental_settings" {
     enable_advanced_database_insights        = bool
     rds_enhanced_monitoring_interval_seconds = number # Enables RDS enhanced monitoring if value is 1, 5, 10, 15, 30 or 60. Disabled if 0
     serve_assets_from_s3                     = optional(bool, false)
+    enable_managed_grafana                   = optional(bool, false)
   })
   validation {
     condition     = contains([0, 1, 5, 10, 15, 30, 60], var.environmental_settings.rds_enhanced_monitoring_interval_seconds)
diff --git a/infra/deployments/forms/tfvars/dev.tfvars b/infra/deployments/forms/tfvars/dev.tfvars
index 147bb6ac4..e93429fe6 100644
--- a/infra/deployments/forms/tfvars/dev.tfvars
+++ b/infra/deployments/forms/tfvars/dev.tfvars
@@ -29,6 +29,7 @@ environmental_settings = {
   enable_advanced_database_insights        = false
   rds_enhanced_monitoring_interval_seconds = 0 # disabled
   serve_assets_from_s3                     = true
+  enable_managed_grafana                   = true
 }
 root_domain = "dev.forms.service.gov.uk"
 additional_dns_records = [
@@ -114,6 +115,7 @@ forms_runner_settings = {
   govuk_one_login_base_url                                        = "https://oidc.integration.account.gov.uk/"
   queue_worker_capacity                                           = 1
   disable_builtin_solidqueue_worker                               = true
+  enable_submission_events_analytics                              = true
 }
 scheduled_smoke_tests_settings = {
   enable_scheduled_smoke_tests = true
diff --git a/infra/deployments/forms/tfvars/production.tfvars b/infra/deployments/forms/tfvars/production.tfvars
index 235d3b44f..6585cb21e 100644
--- a/infra/deployments/forms/tfvars/production.tfvars
+++ b/infra/deployments/forms/tfvars/production.tfvars
@@ -170,6 +170,7 @@ forms_runner_settings = {
   govuk_one_login_base_url                                        = "https://oidc.account.gov.uk/"
   queue_worker_capacity                                           = 6
   disable_builtin_solidqueue_worker                               = true
+  enable_submission_events_analytics                              = false
 }
 scheduled_smoke_tests_settings = {
   enable_scheduled_smoke_tests = true
diff --git a/infra/deployments/forms/tfvars/staging.tfvars b/infra/deployments/forms/tfvars/staging.tfvars
index f142f19b8..962cf4179 100644
--- a/infra/deployments/forms/tfvars/staging.tfvars
+++ b/infra/deployments/forms/tfvars/staging.tfvars
@@ -79,6 +79,7 @@ forms_runner_settings = {
   govuk_one_login_base_url                                        = "https://oidc.integration.account.gov.uk/"
   queue_worker_capacity                                           = 1
   disable_builtin_solidqueue_worker                               = true
+  enable_submission_events_analytics                              = false
 }
 scheduled_smoke_tests_settings = {
   enable_scheduled_smoke_tests = true
diff --git a/infra/modules/deployer-access/policy.tf b/infra/modules/deployer-access/policy.tf
index 3851e2b5e..8fbe86993 100644
--- a/infra/modules/deployer-access/policy.tf
+++ b/infra/modules/deployer-access/policy.tf
@@ -22,6 +22,7 @@ data "aws_iam_policy_document" "forms_infra_1" {
     data.aws_iam_policy_document.ecs.json,
     data.aws_iam_policy_document.elasticache.json,
     data.aws_iam_policy_document.events.json,
+    data.aws_iam_policy_document.grafana.json,
     data.aws_iam_policy_document.guardduty.json,
   ]
 }
@@ -449,6 +450,39 @@ data "aws_iam_policy_document" "events" {
   }
 }
 
+data "aws_iam_policy_document" "grafana" {
+  statement {
+    # CreateWorkspace and the sso/organizations calls do not support resource-level permissions
+    actions = [
+      "grafana:CreateWorkspace",
+      "sso:CreateManagedApplicationInstance",
+      "sso:DeleteManagedApplicationInstance",
+      "sso:GetManagedApplicationInstance",
+      "sso:GetSharedSsoConfiguration",
+      "sso:DescribeRegisteredRegions",
+      "organizations:DescribeOrganization"
+    ]
+    effect    = "Allow"
+    resources = ["*"]
+    sid       = "CreateGrafanaWorkspace"
+  }
+
+  statement {
+    actions = [
+      "grafana:DeleteWorkspace",
+      "grafana:UpdateWorkspace",
+      "grafana:UpdateWorkspaceConfiguration",
+      "grafana:TagResource",
+      "grafana:UntagResource"
+    ]
+    effect = "Allow"
+    resources = [
+      "arn:aws:grafana:eu-west-2:${var.account_id}:/workspaces/*"
+    ]
+    sid = "ManageGrafanaWorkspaces"
+  }
+}
+
 data "aws_iam_policy_document" "guardduty" {
   statement {
     actions = [
@@ -525,7 +559,8 @@ data "aws_iam_policy_document" "iam" {
       "arn:aws:iam::${var.account_id}:role/ecs-events-role",
       "arn:aws:iam::${var.account_id}:role/deployer-${var.environment_name}",
       "arn:aws:iam::${var.account_id}:role/malware-protection-for-s3",
-      "arn:aws:iam::${var.account_id}:role/RDSEnhancedMonitoring"
+      "arn:aws:iam::${var.account_id}:role/RDSEnhancedMonitoring",
+      "arn:aws:iam::${var.account_id}:role/${var.environment_name}-grafana-workspace"
     ]
     sid = "ManageRoles"
   }
diff --git a/infra/modules/submission-events/.gitignore b/infra/modules/submission-events/.gitignore
new file mode 100644
index 000000000..adc61c038
--- /dev/null
+++ b/infra/modules/submission-events/.gitignore
@@ -0,0 +1 @@
+zip-files/*
diff --git a/infra/modules/submission-events/README.md b/infra/modules/submission-events/README.md
new file mode 100644
index 000000000..bb44029a7
--- /dev/null
+++ b/infra/modules/submission-events/README.md
@@ -0,0 +1,44 @@
+# submission-events
+
+Streams `form_submission` log events from the forms-runner CloudWatch log
+group to an Apache Iceberg table in S3 Tables, via a subscription filter and a
+Kinesis Data Firehose stream.
+
+```
+/aws/ecs/forms-runner-<env>
+  → subscription filter { $.event = "form_submission" }
+  → Firehose (transform Lambda unwraps the CloudWatch Logs envelope)
+  → S3 Table s3tablescatalog/govuk-forms-<env>-submission-events, forms.form_submissions
+```
+
+A transform Lambda (lambda/transform.rb) gunzips each CloudWatch Logs record
+and maps its log events onto the table schema as newline-delimited JSON;
+Firehose's built-in decompression processor only supports the S3, Splunk and
+Snowflake destinations, not Iceberg. Firehose invokes the Lambda with batches
+of records (up to 3 MB or 60 seconds of buffered data per invocation), not
+once per event.
+
+The table schema is `submitted_at` (timestamp, renamed from the log event's
+`time` field and normalised to UTC), `form_id` and `form_name` (strings) and
+`preview` (boolean). All other log fields are dropped by the Lambda. Records
+Firehose cannot deliver land in the
+`govuk-forms-<env>-submission-events-errors` bucket.
+
+The table is unpartitioned so that it can be fully managed by Terraform (the
+provider cannot yet set an Iceberg partition spec). At our volumes Iceberg's
+per-file column statistics keep time-range queries cheap regardless.
+
+## Querying
+
+In Athena, select catalog `s3tablescatalog/govuk-forms-<env>-submission-events`
+and database `forms`:
+
+```sql
+SELECT * FROM form_submissions ORDER BY submitted_at DESC LIMIT 10;
+```
+
+## Notes
+
+- CloudWatch Logs sends a control message when the subscription filter is
+  created; the transform Lambda drops it.
+- Preview submissions are included, with `preview = true`.
diff --git a/infra/modules/submission-events/error-bucket.tf b/infra/modules/submission-events/error-bucket.tf
new file mode 100644
index 000000000..39e6f8f2d
--- /dev/null
+++ b/infra/modules/submission-events/error-bucket.tf
@@ -0,0 +1,5 @@
+# Firehose requires an S3 bucket for records it fails to deliver to the table.
+module "error_bucket" {
+  source = "../secure-bucket"
+  name   = "govuk-forms-${var.env_name}-submission-events-errors"
+}
diff --git a/infra/modules/submission-events/firehose.tf b/infra/modules/submission-events/firehose.tf
new file mode 100644
index 000000000..b4b8d7d09
--- /dev/null
+++ b/infra/modules/submission-events/firehose.tf
@@ -0,0 +1,65 @@
+resource "aws_kinesis_firehose_delivery_stream" "submission_events" {
+  name        = local.stream_name
+  destination = "iceberg"
+
+  # No customer-managed key: the data is not sensitive, the AWS-owned key is
+  # free and needs no management.
+  #checkov:skip=CKV_AWS_241:An AWS-owned key is sufficient, no need for CM KMS.
+  server_side_encryption {
+    enabled  = true
+    key_type = "AWS_OWNED_CMK"
+  }
+
+  iceberg_configuration {
+    role_arn           = aws_iam_role.firehose_delivery.arn
+    catalog_arn        = "arn:aws:glue:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:catalog/s3tablescatalog/${local.table_bucket_name}"
+    buffering_interval = var.firehose_buffering_interval_seconds
+
+    destination_table_configuration {
+      database_name = local.namespace_name
+      table_name    = local.table_name
+    }
+
+    s3_configuration {
+      role_arn   = aws_iam_role.firehose_delivery.arn
+      bucket_arn = module.error_bucket.arn
+    }
+
+    # Records arrive as gzipped CloudWatch Logs envelopes. The built-in
+    # Decompression processor is not supported for the Iceberg destination, so
+    # a Lambda unwraps the envelope instead (see lambda/transform.rb).
+    processing_configuration {
+      enabled = true
+
+      processors {
+        type = "Lambda"
+        parameters {
+          parameter_name  = "LambdaArn"
+          parameter_value = "${aws_lambda_function.transform.arn}:$LATEST"
+        }
+      }
+    }
+
+    cloudwatch_logging_options {
+      enabled         = true
+      log_group_name  = aws_cloudwatch_log_group.firehose.name
+      log_stream_name = aws_cloudwatch_log_stream.firehose_delivery.name
+    }
+  }
+
+  # Firehose validates at creation time that the destination table exists and
+  # is reachable through the federated catalog
+  depends_on = [aws_glue_catalog.s3tablescatalog, aws_s3tables_table.form_submissions]
+}
+
+resource "aws_cloudwatch_log_group" "firehose" {
+  #checkov:skip=CKV_AWS_338:We're happy with 30 days retention for now
+  #checkov:skip=CKV_AWS_158:Default AWS SSE is sufficient, no need for CM KMS.
+  name              = "/aws/kinesisfirehose/${local.stream_name}"
+  retention_in_days = 30
+}
+
+resource "aws_cloudwatch_log_stream" "firehose_delivery" {
+  name           = "DestinationDelivery"
+  log_group_name = aws_cloudwatch_log_group.firehose.name
+}
diff --git a/infra/modules/submission-events/glue-catalog.tf b/infra/modules/submission-events/glue-catalog.tf
new file mode 100644
index 000000000..2ec3fb2f8
--- /dev/null
+++ b/infra/modules/submission-events/glue-catalog.tf
@@ -0,0 +1,30 @@
+# Integrates S3 Tables with AWS analytics services (Athena, Firehose, etc.) by
+# federating all S3 table buckets in the account into the Glue Data Catalog.
+# This is an account+region singleton: the name must be "s3tablescatalog" and
+# it covers every table bucket in the account, not just the one in this module.
+# Access control is IAM-based (IAM_ALLOWED_PRINCIPALS), so consumers need IAM
+# permissions only, no Lake Formation grants.
+resource "aws_glue_catalog" "s3tablescatalog" {
+  name = "s3tablescatalog"
+
+  federated_catalog {
+    connection_name = "aws:s3tables"
+    identifier      = "arn:aws:s3tables:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:bucket/*"
+  }
+
+  create_database_default_permissions {
+    permissions = ["ALL"]
+    principal {
+      data_lake_principal_identifier = "IAM_ALLOWED_PRINCIPALS"
+    }
+  }
+
+  create_table_default_permissions {
+    permissions = ["ALL"]
+    principal {
+      data_lake_principal_identifier = "IAM_ALLOWED_PRINCIPALS"
+    }
+  }
+
+  allow_full_table_external_data_access = "True"
+}
diff --git a/infra/modules/submission-events/iam.tf b/infra/modules/submission-events/iam.tf
new file mode 100644
index 000000000..b8ad6fafd
--- /dev/null
+++ b/infra/modules/submission-events/iam.tf
@@ -0,0 +1,145 @@
+# Role assumed by Firehose to write to the S3 Table and the error bucket
+resource "aws_iam_role" "firehose_delivery" {
+  name               = "govuk-forms-submission-events-firehose-${var.env_name}"
+  assume_role_policy = data.aws_iam_policy_document.firehose_assume_role.json
+}
+
+data "aws_iam_policy_document" "firehose_assume_role" {
+  statement {
+    effect  = "Allow"
+    actions = ["sts:AssumeRole"]
+
+    principals {
+      type        = "Service"
+      identifiers = ["firehose.amazonaws.com"]
+    }
+
+    condition {
+      test     = "StringEquals"
+      variable = "sts:ExternalId"
+      values   = [data.aws_caller_identity.current.account_id]
+    }
+  }
+}
+
+resource "aws_iam_role_policy" "firehose_delivery" {
+  name   = "deliver-submission-events"
+  role   = aws_iam_role.firehose_delivery.id
+  policy = data.aws_iam_policy_document.firehose_delivery.json
+}
+
+data "aws_iam_policy_document" "firehose_delivery" {
+  statement {
+    sid    = "ReadAndWriteTableData"
+    effect = "Allow"
+    actions = [
+      "s3tables:GetTableBucket",
+      "s3tables:GetNamespace",
+      "s3tables:GetTable",
+      "s3tables:GetTableData",
+      "s3tables:GetTableMetadataLocation",
+      "s3tables:PutTableData",
+      "s3tables:UpdateTableMetadataLocation"
+    ]
+    resources = [
+      aws_s3tables_table_bucket.submission_events.arn,
+      "${aws_s3tables_table_bucket.submission_events.arn}/table/*"
+    ]
+  }
+
+  statement {
+    sid    = "AccessTableThroughGlueCatalog"
+    effect = "Allow"
+    actions = [
+      "glue:GetDatabase",
+      "glue:GetDatabases",
+      "glue:GetTable",
+      "glue:GetTables",
+      "glue:UpdateTable"
+    ]
+    # Databases and tables in the federated catalog have the catalog path
+    # embedded in their ARNs (e.g. table/s3tablescatalog/<bucket>/<ns>/<table>)
+    resources = [
+      "arn:aws:glue:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:catalog",
+      "arn:aws:glue:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:catalog/s3tablescatalog",
+      "arn:aws:glue:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:catalog/s3tablescatalog/${local.table_bucket_name}",
+      "arn:aws:glue:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:database/s3tablescatalog/${local.table_bucket_name}/${local.namespace_name}",
+      "arn:aws:glue:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:table/s3tablescatalog/${local.table_bucket_name}/${local.namespace_name}/*"
+    ]
+  }
+
+  statement {
+    sid    = "WriteUndeliveredRecordsToErrorBucket"
+    effect = "Allow"
+    actions = [
+      "s3:AbortMultipartUpload",
+      "s3:GetBucketLocation",
+      "s3:GetObject",
+      "s3:ListBucket",
+      "s3:ListBucketMultipartUploads",
+      "s3:PutObject"
+    ]
+    resources = [
+      module.error_bucket.arn,
+      "${module.error_bucket.arn}/*"
+    ]
+  }
+
+  statement {
+    sid    = "InvokeTransformLambda"
+    effect = "Allow"
+    actions = [
+      "lambda:InvokeFunction",
+      "lambda:GetFunctionConfiguration"
+    ]
+    resources = ["${aws_lambda_function.transform.arn}:$LATEST"]
+  }
+
+  statement {
+    sid       = "WriteDeliveryErrorLogs"
+    effect    = "Allow"
+    actions   = ["logs:PutLogEvents"]
+    resources = ["${aws_cloudwatch_log_group.firehose.arn}:log-stream:*"]
+  }
+}
+
+# Role assumed by CloudWatch Logs to put log events onto the Firehose stream
+resource "aws_iam_role" "log_subscription" {
+  name               = "govuk-forms-submission-events-subscription-${var.env_name}"
+  assume_role_policy = data.aws_iam_policy_document.log_subscription_assume_role.json
+}
+
+data "aws_iam_policy_document" "log_subscription_assume_role" {
+  statement {
+    effect  = "Allow"
+    actions = ["sts:AssumeRole"]
+
+    principals {
+      type        = "Service"
+      identifiers = ["logs.amazonaws.com"]
+    }
+
+    condition {
+      test     = "StringLike"
+      variable = "aws:SourceArn"
+      values   = ["arn:aws:logs:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:*"]
+    }
+  }
+}
+
+resource "aws_iam_role_policy" "log_subscription" {
+  name   = "put-records-to-firehose"
+  role   = aws_iam_role.log_subscription.id
+  policy = data.aws_iam_policy_document.log_subscription.json
+}
+
+data "aws_iam_policy_document" "log_subscription" {
+  statement {
+    effect = "Allow"
+    actions = [
+      "firehose:PutRecord",
+      "firehose:PutRecordBatch"
+    ]
+    resources = [aws_kinesis_firehose_delivery_stream.submission_events.arn]
+  }
+}
diff --git a/infra/modules/submission-events/lambda.tf b/infra/modules/submission-events/lambda.tf
new file mode 100644
index 000000000..84a97ba8d
--- /dev/null
+++ b/infra/modules/submission-events/lambda.tf
@@ -0,0 +1,54 @@
+data "archive_file" "transform" {
+  type        = "zip"
+  source_dir  = "${path.module}/lambda"
+  output_path = "${path.module}/zip-files/transform-lambda.zip"
+}
+
+resource "aws_lambda_function" "transform" {
+  #checkov:skip=CKV_AWS_272:we're not doing code signing on this lambda at the moment
+  #checkov:skip=CKV_AWS_116:no dead letter queue, Firehose retries and then writes failures to the error bucket
+  #checkov:skip=CKV_AWS_117:lambda does not need access to things inside the VPC
+  #checkov:skip=CKV_AWS_50 :not using X-Ray
+
+  function_name = "submission-events-transform-${var.env_name}"
+  description   = "Unwraps CloudWatch Logs records for the submission events Firehose stream"
+  role          = aws_iam_role.transform_lambda.arn
+
+  runtime                        = "ruby3.4"
+  handler                        = "transform.handler"
+  reserved_concurrent_executions = 10
+  timeout                        = 120
+  memory_size                    = 256
+
+  filename         = data.archive_file.transform.output_path
+  source_code_hash = data.archive_file.transform.output_base64sha256
+}
+
+resource "aws_cloudwatch_log_group" "transform_lambda" {
+  #checkov:skip=CKV_AWS_338:We're happy with 30 days retention for now
+  #checkov:skip=CKV_AWS_158:Amazon managed SSE is sufficient.
+  name              = "/aws/lambda/${aws_lambda_function.transform.function_name}"
+  retention_in_days = 30
+}
+
+resource "aws_iam_role" "transform_lambda" {
+  name               = "govuk-forms-submission-events-transform-${var.env_name}"
+  assume_role_policy = data.aws_iam_policy_document.transform_lambda_assume_role.json
+}
+
+data "aws_iam_policy_document" "transform_lambda_assume_role" {
+  statement {
+    effect  = "Allow"
+    actions = ["sts:AssumeRole"]
+
+    principals {
+      type        = "Service"
+      identifiers = ["lambda.amazonaws.com"]
+    }
+  }
+}
+
+resource "aws_iam_role_policy_attachment" "transform_lambda_basic" {
+  role       = aws_iam_role.transform_lambda.name
+  policy_arn = "arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole"
+}
diff --git a/infra/modules/submission-events/lambda/transform.rb b/infra/modules/submission-events/lambda/transform.rb
new file mode 100644
index 000000000..8c9dc313a
--- /dev/null
+++ b/infra/modules/submission-events/lambda/transform.rb
@@ -0,0 +1,39 @@
+require "base64"
+require "json"
+require "stringio"
+require "time"
+require "zlib"
+
+# Firehose transform: unwraps the gzipped CloudWatch Logs envelope and maps
+# each log event's message onto the form_submissions table schema, one JSON
+# object per line. Exactly one output record per input record.
+def handler(event:, context:)
+  records = event["records"].map do |record|
+    begin
+      envelope = JSON.parse(Zlib::GzipReader.new(StringIO.new(Base64.decode64(record["data"]))).read)
+
+      if envelope["messageType"] == "CONTROL_MESSAGE"
+        { "recordId" => record["recordId"], "result" => "Dropped" }
+      else
+        lines = envelope["logEvents"].map do |log_event|
+          message = JSON.parse(log_event["message"])
+          {
+            "submitted_at" => Time.iso8601(message.fetch("time")).utc.strftime("%Y-%m-%dT%H:%M:%S.%LZ"),
+            "form_id" => message["form_id"],
+            "form_name" => message["form_name"],
+            "preview" => message["preview"] == "true"
+          }.to_json
+        end
+        {
+          "recordId" => record["recordId"],
+          "result" => "Ok",
+          "data" => Base64.strict_encode64(lines.join("\n"))
+        }
+      end
+    rescue JSON::ParserError, KeyError, ArgumentError, Zlib::GzipFile::Error
+      { "recordId" => record["recordId"], "result" => "ProcessingFailed", "data" => record["data"] }
+    end
+  end
+
+  { "records" => records }
+end
diff --git a/infra/modules/submission-events/main.tf b/infra/modules/submission-events/main.tf
new file mode 100644
index 000000000..124e80d2d
--- /dev/null
+++ b/infra/modules/submission-events/main.tf
@@ -0,0 +1,17 @@
+locals {
+  # Created by the ecs-service module, see infra/modules/ecs-service/ecs.tf
+  log_group_name = "/aws/ecs/forms-runner-${var.env_name}"
+
+  table_bucket_name = "govuk-forms-${var.env_name}-submission-events"
+  namespace_name    = "forms"
+  table_name        = "form_submissions"
+  stream_name       = "submission-events-${var.env_name}"
+}
+
+data "aws_caller_identity" "current" {}
+
+data "aws_region" "current" {}
+
+data "aws_cloudwatch_log_group" "forms_runner" {
+  name = local.log_group_name
+}
diff --git a/infra/modules/submission-events/outputs.tf b/infra/modules/submission-events/outputs.tf
new file mode 100644
index 000000000..1f6f0d41e
--- /dev/null
+++ b/infra/modules/submission-events/outputs.tf
@@ -0,0 +1,15 @@
+output "firehose_stream_arn" {
+  value = aws_kinesis_firehose_delivery_stream.submission_events.arn
+}
+
+output "table_bucket_name" {
+  value = aws_s3tables_table_bucket.submission_events.name
+}
+
+output "table_bucket_arn" {
+  value = aws_s3tables_table_bucket.submission_events.arn
+}
+
+output "error_bucket_name" {
+  value = module.error_bucket.name
+}
diff --git a/infra/modules/submission-events/s3-tables.tf b/infra/modules/submission-events/s3-tables.tf
new file mode 100644
index 000000000..eba4e0c6a
--- /dev/null
+++ b/infra/modules/submission-events/s3-tables.tf
@@ -0,0 +1,43 @@
+resource "aws_s3tables_table_bucket" "submission_events" {
+  name = local.table_bucket_name
+}
+
+resource "aws_s3tables_namespace" "forms" {
+  namespace        = local.namespace_name
+  table_bucket_arn = aws_s3tables_table_bucket.submission_events.arn
+}
+
+# The table is deliberately unpartitioned: at our data volumes Iceberg's
+# per-file column statistics prune time-range queries well enough, and an
+# unpartitioned table can be fully managed here (the provider cannot yet set a
+# partition spec, https://github.com/hashicorp/terraform-provider-aws/issues/46494).
+resource "aws_s3tables_table" "form_submissions" {
+  name             = local.table_name
+  namespace        = aws_s3tables_namespace.forms.namespace
+  table_bucket_arn = aws_s3tables_table_bucket.submission_events.arn
+  format           = "ICEBERG"
+
+  metadata {
+    iceberg {
+      schema {
+        field {
+          name     = "submitted_at"
+          type     = "timestamp"
+          required = true
+        }
+        field {
+          name = "form_id"
+          type = "string"
+        }
+        field {
+          name = "form_name"
+          type = "string"
+        }
+        field {
+          name = "preview"
+          type = "boolean"
+        }
+      }
+    }
+  }
+}
diff --git a/infra/modules/submission-events/subscription-filter.tf b/infra/modules/submission-events/subscription-filter.tf
new file mode 100644
index 000000000..7b8af6d7f
--- /dev/null
+++ b/infra/modules/submission-events/subscription-filter.tf
@@ -0,0 +1,12 @@
+# This uses the second and final subscription filter slot on the forms-runner
+# log group (CloudWatch Logs allows two per log group; the first is
+# via-cribl-to-splunk in infra/modules/ecs-service/logging.tf).
+resource "aws_cloudwatch_log_subscription_filter" "submission_events" {
+  name = "submission-events-to-firehose"
+
+  log_group_name = data.aws_cloudwatch_log_group.forms_runner.name
+
+  filter_pattern  = "{ $.event = \"form_submission\" }"
+  destination_arn = aws_kinesis_firehose_delivery_stream.submission_events.arn
+  role_arn        = aws_iam_role.log_subscription.arn
+}
diff --git a/infra/modules/submission-events/variables.tf b/infra/modules/submission-events/variables.tf
new file mode 100644
index 000000000..30af652d4
--- /dev/null
+++ b/infra/modules/submission-events/variables.tf
@@ -0,0 +1,10 @@
+variable "env_name" {
+  type        = string
+  description = "The name of the environment"
+}
+
+variable "firehose_buffering_interval_seconds" {
+  type        = number
+  description = "How long Firehose buffers records before committing them to the table"
+  default     = 300
+}