Fix: let the database enforce the events -> sessions cascade in DatabaseSessionService

core/src/sessions/database_session_service.ts

@@ -481,9 +481,7 @@ export class DatabaseSessionService extends BaseSessionService {
       } else {
         const newStorageEvent = txEm.create(StorageEvent, {
           id: trimmedEvent.id,
-          appName: session.appName,
-          userId: session.userId,
-          sessionId: session.id,
+          session: storageSession,
           invocationId: trimmedEvent.invocationId,
           timestamp: new Date(trimmedEvent.timestamp),
           eventData: trimmedEvent,

core/src/sessions/db/operations.ts

@@ -4,14 +4,27 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-import {MikroORM, Options as MikroORMOptions} from '@mikro-orm/core';
+import {
+  MikroORM,
+  Options as MikroORMOptions,
+  QueryResult,
+} from '@mikro-orm/core';
+import {logger} from '../../utils/logger.js';
 import {
   ENTITIES,
   SCHEMA_VERSION_1_JSON,
   SCHEMA_VERSION_KEY,
   StorageMetadata,
 } from './schema.js';
 
+const DELETE_ORPHANED_EVENTS_SQL = `DELETE FROM events
+WHERE NOT EXISTS (
+  SELECT 1 FROM sessions
+  WHERE sessions.app_name = events.app_name
+    AND sessions.user_id = events.user_id
+    AND sessions.id = events.session_id
+)`;
+
 /**
  * Parses a database connection URI and returns MikroORM Options.
  *
@@ -61,6 +74,21 @@ export async function getConnectionOptionsFromUri(
   } as MikroORMOptions;
 }
 
+/**
+ * Deletes event rows whose session no longer exists.
+ *
+ * @param orm The MikroORM instance.
+ * @returns Promise<void>
+ */
+export async function deleteOrphanedEvents(orm: MikroORM): Promise<void> {
+  const result: QueryResult = await orm.em
+    .getConnection()
+    .execute(DELETE_ORPHANED_EVENTS_SQL, [], 'run');
+  logger.warn(
+    `Deleted ${result.affectedRows} event rows whose session no longer exists.`,
+  );
+}
+
 /**
  * Creates a database and tables if they don't exist.
  *
@@ -71,8 +99,21 @@ export async function ensureDatabaseCreated(orm: MikroORM): Promise<void> {
   // creates database if it doesn't exist
   await orm.schema.ensureDatabase();
 
-  // creates tables if they don't exist. Safe mode prevents dropping columns or tables.
-  await orm.schema.updateSchema({safe: true});
+  try {
+    // creates tables if they don't exist. Safe mode prevents dropping columns or tables.
+    await orm.schema.updateSchema({safe: true});
+  } catch (error) {
+    // A database created before the events -> sessions foreign key can hold
+    // event rows whose session is already gone, and the constraint cannot be
+    // added while they exist. Those rows are unreachable through the service,
+    // so drop them and let the schema update finish.
+    logger.warn(
+      'Schema update failed; retrying after deleting orphaned events.',
+      error,
+    );
+    await deleteOrphanedEvents(orm);
+    await orm.schema.updateSchema({safe: true});
+  }
 }
 
 /**

core/src/sessions/db/schema.ts

@@ -4,7 +4,13 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-import {Entity, JsonType, PrimaryKey, Property} from '@mikro-orm/core';
+import {
+  Entity,
+  JsonType,
+  ManyToOne,
+  PrimaryKey,
+  Property,
+} from '@mikro-orm/core';
 import {
   Event,
   transformToCamelCaseEvent,
@@ -96,11 +102,14 @@ export class StorageUserState {
   [PrimaryKey.name]?: [string, string];
 }
 
+/**
+ * The primary key is declared in the order `app_name, user_id, id` to match
+ * adk-python's v1 schema. MikroORM binds a composite foreign key in the
+ * target's declaration order, so this order is what keeps the `events` columns
+ * laid out as `id, app_name, user_id, session_id`.
+ */
 @Entity({tableName: 'sessions'})
 export class StorageSession {
-  @PrimaryKey({type: 'string', length: STORAGE_KEY_COLUMN_LENGTH})
-  id!: string;
-
   @PrimaryKey({
     type: 'string',
     fieldName: 'app_name',
@@ -115,6 +124,9 @@ export class StorageSession {
   })
   userId!: string;
 
+  @PrimaryKey({type: 'string', length: STORAGE_KEY_COLUMN_LENGTH})
+  id!: string;
+
   @Property({type: 'json'})
   state!: Record<string, unknown>;
 
@@ -135,30 +147,33 @@ export class StorageSession {
   [PrimaryKey.name]?: [string, string, string];
 }
 
+/**
+ * The `session` relation owns `app_name`, `user_id` and `session_id`, so the
+ * database enforces that every event belongs to a live session and removes the
+ * events when that session row is deleted. `appName`, `userId` and `sessionId`
+ * stay declared as non-persistent mirrors of the same columns so that queries
+ * can filter on them without loading the relation.
+ */
 @Entity({tableName: 'events'})
 export class StorageEvent {
   @PrimaryKey({type: 'string', length: STORAGE_KEY_COLUMN_LENGTH})
   id!: string;
 
-  @PrimaryKey({
-    type: 'string',
-    fieldName: 'app_name',
-    length: STORAGE_KEY_COLUMN_LENGTH,
+  @ManyToOne(() => StorageSession, {
+    primary: true,
+    joinColumns: ['app_name', 'user_id', 'session_id'],
+    deleteRule: 'cascade',
+    updateRule: 'cascade',
   })
+  session!: StorageSession;
+
+  @Property({type: 'string', fieldName: 'app_name', persist: false})
   appName!: string;
 
-  @PrimaryKey({
-    type: 'string',
-    fieldName: 'user_id',
-    length: STORAGE_KEY_COLUMN_LENGTH,
-  })
+  @Property({type: 'string', fieldName: 'user_id', persist: false})
   userId!: string;
 
-  @PrimaryKey({
-    type: 'string',
-    fieldName: 'session_id',
-    length: STORAGE_KEY_COLUMN_LENGTH,
-  })
+  @Property({type: 'string', fieldName: 'session_id', persist: false})
   sessionId!: string;
 
   @Property({type: 'string', fieldName: 'invocation_id'})