fix(chat,mux): atomic message insert, fix mislabeled health field, real recording toggle

app/api/streams/chat/route.ts

@@ -132,26 +132,31 @@ export async function POST(req: NextRequest) {
       );
     }
 
+    // Single atomic statement: the INSERT and the counter increment are in the
+    // same implicit transaction, so a crash or connection drop between them
+    // can never leave total_messages permanently out of sync (#764).
     const messageResult = await sql`
-      INSERT INTO chat_messages (
-        user_id,
-        username,
-        stream_session_id,
-        content,
-        message_type,
-        created_at
+      WITH new_msg AS (
+        INSERT INTO chat_messages (
+          user_id,
+          username,
+          stream_session_id,
+          content,
+          message_type,
+          created_at
+        )
+        VALUES (${sender_id}, ${sender_username}, ${session_id}, ${content}, ${messageType}, CURRENT_TIMESTAMP)
+        RETURNING id, created_at, stream_session_id
+      ),
+      _counter AS (
+        UPDATE stream_sessions
+        SET total_messages = total_messages + 1
+        WHERE id = (SELECT stream_session_id FROM new_msg)
       )
-      VALUES (${sender_id}, ${sender_username}, ${session_id}, ${content}, ${messageType}, CURRENT_TIMESTAMP)
-      RETURNING id, created_at
+      SELECT id, created_at FROM new_msg
     `;
 
     const newMessage = messageResult.rows[0];
-
-    await sql`
-      UPDATE stream_sessions SET
-        total_messages = total_messages + 1
-      WHERE id = ${session_id}
-    `;
     await chatWindowCache.invalidate([chatTag(playbackId)]);
 
     const formattedMessage = {

lib/mux/server.ts

@@ -248,23 +248,61 @@ export async function getPlaybackUrl(playbackId: string) {
 }
 
 /**
- * Update recording preference on an existing Mux live stream.
- * Must be called whenever enable_recording changes — recording is baked into
- * new_asset_settings at stream creation time, so toggling the DB flag alone
- * has no effect on streams that already exist in Mux.
+ * Update the recording preference on an existing Mux live stream.
+ *
+ * Mux allows updating `new_asset_settings` via the live-stream update API.
+ * Enabling sets a public-playback asset policy so the stream is saved when it
+ * ends; disabling clears it so no new asset is created. Note: changes take
+ * effect for the *next* broadcast segment — Mux cannot retroactively stop
+ * recording a segment already in progress, so the UI should inform creators
+ * accordingly (#755).
  */
-/**
- * Recording policy (playback_policy inside new_asset_settings) is baked in at
- * stream-creation time and cannot be changed via the Mux update endpoint.
- * This function is a no-op intentionally — recording changes take effect the
- * next time a Mux stream is created for this user.
- */
-
 export async function updateMuxStreamRecording(
-  _streamId: string,
-  _enable: boolean
-): Promise<{ success: true }> {
-  return { success: true };
+  streamId: string,
+  enable: boolean
+): Promise<{ success: boolean; streamId: string }> {
+  const startTime = Date.now();
+  try {
+    logger.info("Updating Mux stream recording setting", {
+      operation: "updateMuxStreamRecording",
+      streamId,
+      enable,
+    });
+
+    await muxCall(options =>
+      mux.video.liveStreams.update(
+        streamId,
+        {
+          new_asset_settings: enable
+            ? { playback_policy: ["public"] as any }
+            : {},
+        },
+        options
+      )
+    );
+
+    const durationMs = Date.now() - startTime;
+    logger.info("Mux stream recording updated", {
+      operation: "updateMuxStreamRecording",
+      streamId,
+      enable,
+      durationMs,
+    });
+
+    return { success: true, streamId };
+  } catch (error) {
+    const durationMs = Date.now() - startTime;
+    logger.error("Mux recording update failed", {
+      operation: "updateMuxStreamRecording",
+      streamId,
+      enable,
+      durationMs,
+      errorMessage: error instanceof Error ? error.message : String(error),
+    });
+    throw new Error(
+      `Failed to update Mux recording setting: ${error instanceof Error ? error.message : String(error)}`
+    );
+  }
 }
 
 export async function enableMuxStreamRecording(streamId: string) {
@@ -353,7 +391,11 @@ export async function getMuxStreamHealth(streamId: string) {
       status: liveStream.status,
       reconnectWindow: liveStream.reconnect_window,
       latencyMode: liveStream.latency_mode,
-      lastSeen: liveStream.recent_asset_ids?.[0] || null,
+      // recent_asset_ids contains Mux asset UUIDs, not timestamps (#759).
+      // Exposed as recentAssetId so callers are not tempted to pass it to
+      // `new Date()`. Use createdAt for time-based recency.
+      recentAssetId: liveStream.recent_asset_ids?.[0] ?? null,
+      createdAt: liveStream.created_at ?? null,
     };
   } catch (error) {
     console.error("Mux stream health error:", error);