[#102] Use UTC explicitly in all event timestamps

Changelog: fix

Author: James Youngblood

.gitlab/ci/scripts/integration-moderation.sh

@@ -39,6 +39,11 @@ cleanup() {
 }
 trap cleanup EXIT
 
+# Clear any stack left behind by a previous job that was hard-killed before its
+# cleanup trap could run (a stale broker/volume would otherwise be reused).
+echo "==> Clearing any stale stack…"
+docker compose down -v >/dev/null 2>&1 || true
+
 echo "==> Bringing up the server stack…"
 docker compose up -d --build --wait postgres rustfs kafka server
 

.gitlab/ci/test_moderation.yml

@@ -9,6 +9,9 @@
 .moderation-integration-workflow:
   image: $RUST_DIND_IMAGE
   stage: test
+  # Enforce serial execution of moderation integration tests using a `resource_group`, so that the
+  # Docker stack of one CI runner doing the test doesn't interfere with that of the other.
+  resource_group: moderation-integration
   rules:
     - if: $CI_COMMIT_TAG =~ /^moderation-/
       when: always

services/common/kafka/src/config.rs

@@ -4,31 +4,30 @@ use rdkafka::config::ClientConfig;
 
 /// Apply the broker, security protocol, and SASL settings shared by
 /// every client to `config`.
+///
+/// SASL properties (`sasl.mechanism`, `sasl.username`, `sasl.password`) are
+/// only set when their corresponding environment variables are explicitly
+/// configured.
 pub(crate) fn set_defaults(config: &mut ClientConfig) {
     let brokers =
         env::var("POLYCENTRIC_KAFKA_BROKERS").unwrap_or_else(|_| "localhost:9092".to_string());
-    config
-        .set("bootstrap.servers", brokers)
-        .set(
-            "security.protocol",
-            env::var("POLYCENTRIC_KAFKA_SECURITY_PROTOCOL")
-                .unwrap_or_else(|_| "PLAINTEXT".to_string()),
-        )
-        .set(
-            "sasl.mechanism",
-            env::var("POLYCENTRIC_KAFKA_SASL_MECHANISM").unwrap_or_else(|_| "PLAIN".to_string()),
-        )
-        .set(
-            "sasl.username",
-            env::var("POLYCENTRIC_KAFKA_SASL_USERNAME").unwrap_or_default(),
-        )
-        .set(
-            "sasl.password",
-            env::var("POLYCENTRIC_KAFKA_SASL_PASSWORD").unwrap_or_default(),
-        )
-        .set(
-            "broker.address.family",
-            env::var("POLYCENTRIC_KAFKA_BROKER_ADDRESS_FAMILY")
-                .unwrap_or_else(|_| "any".to_string()),
-        );
+    config.set("bootstrap.servers", brokers).set(
+        "security.protocol",
+        env::var("POLYCENTRIC_KAFKA_SECURITY_PROTOCOL").unwrap_or_else(|_| "PLAINTEXT".to_string()),
+    );
+
+    if let Ok(value) = env::var("POLYCENTRIC_KAFKA_SASL_MECHANISM") {
+        config.set("sasl.mechanism", value);
+    }
+    if let Ok(value) = env::var("POLYCENTRIC_KAFKA_SASL_USERNAME") {
+        config.set("sasl.username", value);
+    }
+    if let Ok(value) = env::var("POLYCENTRIC_KAFKA_SASL_PASSWORD") {
+        config.set("sasl.password", value);
+    }
+
+    config.set(
+        "broker.address.family",
+        env::var("POLYCENTRIC_KAFKA_BROKER_ADDRESS_FAMILY").unwrap_or_else(|_| "any".to_string()),
+    );
 }

services/common/kafka/src/consumer.rs

@@ -14,12 +14,21 @@ impl ClientContext for CustomContext {}
 impl ConsumerContext for CustomContext {}
 
 /// Build a subscribed `StreamConsumer` with auto-commit disabled.
+///
+/// Override the Kafka `auto.offset.reset` property using the
+/// `POLYCENTRIC_KAFKA_AUTO_OFFSET_RESET` environment variable
+/// (used in moderation integration test).
 pub async fn build_consumer(group_id: &str, topics: &[&str]) -> StreamConsumer<CustomContext> {
     let mut config = ClientConfig::new();
     set_defaults(&mut config);
     config
         .set("group.id", group_id)
-        .set("enable.auto.commit", "false");
+        .set("enable.auto.commit", "false")
+        .set(
+            "auto.offset.reset",
+            std::env::var("POLYCENTRIC_KAFKA_AUTO_OFFSET_RESET")
+                .unwrap_or_else(|_| "latest".to_string()),
+        );
 
     let consumer: StreamConsumer<CustomContext> = config
         .create_with_context(CustomContext)

services/moderation/entity/src/created_content_model.rs

@@ -16,7 +16,7 @@ pub struct Model {
     // Serialized `Content` proto (the bytes the digest is taken over).
     pub serialized_bytes: Vec<u8>,
 
-    pub created_at: TimeDateTime,
+    pub created_at: TimeDateTimeWithTimeZone,
 }
 
 impl ActiveModelBehavior for ActiveModel {}

services/moderation/entity/src/created_event_model.rs

@@ -37,7 +37,7 @@ pub struct Model {
     pub event_bytes: Vec<u8>,
 
     // Timestamp the event was created (from the event's created_at).
-    pub created_at: TimeDateTime,
+    pub created_at: TimeDateTimeWithTimeZone,
 }
 
 impl ActiveModelBehavior for ActiveModel {}

services/moderation/entity/src/processed_content_model.rs

@@ -21,8 +21,8 @@ pub struct Model {
     #[sea_orm(primary_key)]
     pub digest_bytes: Vec<u8>,
 
-    pub created_at: TimeDateTime,
-    pub updated_at: TimeDateTime,
+    pub created_at: TimeDateTimeWithTimeZone,
+    pub updated_at: TimeDateTimeWithTimeZone,
 
     pub status: Status,
 

services/moderation/migration/src/lib.rs

@@ -2,6 +2,7 @@ pub use sea_orm_migration::prelude::*;
 
 mod m20220101_000001_create_table;
 mod m20260604_000001_create_created_tables;
+mod m20260618_000001_timestamps_to_timestamptz;
 
 pub struct Migrator;
 
@@ -11,6 +12,7 @@ impl MigratorTrait for Migrator {
         vec![
             Box::new(m20220101_000001_create_table::Migration),
             Box::new(m20260604_000001_create_created_tables::Migration),
+            Box::new(m20260618_000001_timestamps_to_timestamptz::Migration),
         ]
     }
 }

services/moderation/migration/src/m20260618_000001_timestamps_to_timestamptz.rs

@@ -0,0 +1,68 @@
+use sea_orm_migration::prelude::*;
+use sea_orm_migration::sea_orm::ConnectionTrait;
+
+/// Convert the moderation timestamp columns from `timestamp without time zone`
+/// (naive) to `timestamptz`, so stored instants are unambiguous UTC and can't
+/// be reinterpreted in a reader's local timezone (issue #102).
+///
+/// Each conversion is guarded on the column's current type so it is a no-op on
+/// a fresh database (where the entity-derived `created_*` tables are already
+/// `timestamptz`) and only rewrites legacy naive columns on existing
+/// deployments. Existing naive values are stamped as UTC via `AT TIME ZONE
+/// 'UTC'` rather than relying on the session timezone.
+#[derive(DeriveMigrationName)]
+pub struct Migration;
+
+const COLUMNS: &[(&str, &str)] = &[
+    ("processed_content", "created_at"),
+    ("processed_content", "updated_at"),
+    ("created_content", "created_at"),
+    ("created_event", "created_at"),
+];
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        let conn = manager.get_connection();
+        for (table, column) in COLUMNS {
+            conn.execute_unprepared(&format!(
+                r#"DO $$
+BEGIN
+  IF EXISTS (
+    SELECT 1 FROM information_schema.columns
+    WHERE table_name = '{table}' AND column_name = '{column}'
+      AND table_schema = current_schema()
+      AND data_type = 'timestamp without time zone'
+  ) THEN
+    ALTER TABLE "{table}"
+      ALTER COLUMN "{column}" TYPE timestamptz USING "{column}" AT TIME ZONE 'UTC';
+  END IF;
+END $$;"#
+            ))
+            .await?;
+        }
+        Ok(())
+    }
+
+    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        let conn = manager.get_connection();
+        for (table, column) in COLUMNS {
+            conn.execute_unprepared(&format!(
+                r#"DO $$
+BEGIN
+  IF EXISTS (
+    SELECT 1 FROM information_schema.columns
+    WHERE table_name = '{table}' AND column_name = '{column}'
+      AND table_schema = current_schema()
+      AND data_type = 'timestamp with time zone'
+  ) THEN
+    ALTER TABLE "{table}"
+      ALTER COLUMN "{column}" TYPE timestamp USING "{column}" AT TIME ZONE 'UTC';
+  END IF;
+END $$;"#
+            ))
+            .await?;
+        }
+        Ok(())
+    }
+}

services/moderation/src/db.rs

@@ -9,7 +9,7 @@ pub async fn connect() -> Result<DatabaseConnection, DbErr> {
     let schema = std::env::var("POLYCENTRIC_MODERATION_DATABASE_SCHEMA")
         .unwrap_or_else(|_| "moderation".to_string());
 
-    let mut opt = ConnectOptions::new(database_url);
+    let mut opt = ConnectOptions::new(with_utc_timezone(&database_url));
     opt.set_schema_search_path(&schema);
     let connection = Database::connect(opt).await?;
 
@@ -24,3 +24,13 @@ pub async fn run_migrations(connection: &DatabaseConnection) -> Result<(), DbErr
     Migrator::up(connection, None).await?;
     Ok(())
 }
+
+/// Strictly enforce a UTC timezone connection by appending the
+/// `options=-c timezone=UTC` parameter.
+fn with_utc_timezone(url: &str) -> String {
+    if url.contains("timezone") {
+        return url.to_string();
+    }
+    let sep = if url.contains('?') { '&' } else { '?' };
+    format!("{url}{sep}options=-c%20timezone%3DUTC")
+}