Merge branch 'dev' into feat/ip-filtering

server/db/pg/schema/privateSchema.ts

@@ -468,6 +468,9 @@ export const eventStreamingDestinations = pgTable(
         sendRequestLogs: boolean("sendRequestLogs").notNull().default(false),
         sendActionLogs: boolean("sendActionLogs").notNull().default(false),
         sendAccessLogs: boolean("sendAccessLogs").notNull().default(false),
+        sendAISessionLogs: boolean("sendAISessionLogs")
+            .notNull()
+            .default(false),
         type: varchar("type", { length: 50 }).notNull(), // e.g. "http", "kafka", etc.
         config: text("config").notNull(), // JSON string with the configuration for the destination
         enabled: boolean("enabled").notNull().default(true),

server/db/sqlite/schema/privateSchema.ts

@@ -459,6 +459,9 @@ export const eventStreamingDestinations = sqliteTable(
         sendAccessLogs: integer("sendAccessLogs", { mode: "boolean" })
             .notNull()
             .default(false),
+        sendAISessionLogs: integer("sendAISessionLogs", { mode: "boolean" })
+            .notNull()
+            .default(false),
         type: text("type").notNull(), // e.g. "http", "kafka", etc.
         config: text("config").notNull(), // JSON string with the configuration for the destination
         enabled: integer("enabled", { mode: "boolean" })

server/lib/exitNodes/subnet.ts

@@ -1,20 +1,26 @@
-import { db, exitNodes, Transaction } from "@server/db";
+import { db, exitNodes, exitNodeOrgs, Transaction } from "@server/db";
 import config from "@server/lib/config";
 import { findNextAvailableCidr } from "@server/lib/ip";
 import { lockManager } from "#dynamic/lib/lock";
+import { eq } from "drizzle-orm";
 
 /**
  * Reserves the next available exit node subnet.
  *
- * Exit node subnets must never overlap with one another - regardless of
- * which org(s) they belong to - since HA exit nodes can end up routing for
- * the same org. This acquires a lock that the caller MUST release (via the
- * returned `release`) only after the chosen address has been durably
- * persisted (e.g. after the enclosing transaction commits), otherwise
- * concurrent callers can race and pick the same subnet.
+ * There isn't enough address space to give every exit node in every org a
+ * globally unique subnet, so we only guarantee uniqueness among exit nodes
+ * that already belong to the same org - that's all that actually matters,
+ * since HA only routes multiple exit nodes for a single org. Pass `orgId` to
+ * scope the search to that org's existing exit nodes; without it, the search
+ * considers every exit node (used by flows with no org context, e.g. the
+ * initial gerbil exit node bootstrap). This acquires a lock that the caller
+ * MUST release (via the returned `release`) only after the chosen address
+ * has been durably persisted (e.g. after the enclosing transaction commits),
+ * otherwise concurrent callers can race and pick the same subnet.
  */
 export async function getNextAvailableSubnet(
-    trx: Transaction | typeof db = db
+    trx: Transaction | typeof db = db,
+    orgId?: string
 ): Promise<{ value: string; release: () => Promise<void> }> {
     const lockKey = "exit-node-subnet-allocation";
     const acquired = await lockManager.acquireLockWithRetry(lockKey, 6000);
@@ -24,12 +30,19 @@ export async function getNextAvailableSubnet(
     const release = () => lockManager.releaseLock(lockKey, acquired);
 
     try {
-        // Get all existing subnets from routes table
-        const existingAddresses = await trx
-            .select({
-                address: exitNodes.address
-            })
-            .from(exitNodes);
+        // Get existing subnets, scoped to this org's exit nodes when known
+        const existingAddresses = orgId
+            ? await trx
+                  .select({ address: exitNodes.address })
+                  .from(exitNodes)
+                  .innerJoin(
+                      exitNodeOrgs,
+                      eq(exitNodeOrgs.exitNodeId, exitNodes.exitNodeId)
+                  )
+                  .where(eq(exitNodeOrgs.orgId, orgId))
+            : await trx
+                  .select({ address: exitNodes.address })
+                  .from(exitNodes);
 
         const addresses = existingAddresses.map((a) => a.address);
         let subnet = findNextAvailableCidr(

server/private/lib/logStreaming/LogStreamingManager.ts

@@ -19,7 +19,8 @@ import {
     requestAuditLog,
     actionAuditLog,
     accessAuditLog,
-    connectionAuditLog
+    connectionAuditLog,
+    aiSessionLog
 } from "@server/db";
 import logger from "@server/logger";
 import { and, eq, gt, desc, max, sql } from "drizzle-orm";
@@ -309,6 +310,7 @@ export class LogStreamingManager {
         if (dest.sendActionLogs) enabledTypes.push("action");
         if (dest.sendAccessLogs) enabledTypes.push("access");
         if (dest.sendConnectionLogs) enabledTypes.push("connection");
+        if (dest.sendAISessionLogs) enabledTypes.push("aiSession");
 
         if (enabledTypes.length === 0) return;
 
@@ -585,6 +587,13 @@ export class LogStreamingManager {
                         .where(eq(connectionAuditLog.orgId, orgId));
                     return row?.maxId ?? 0;
                 }
+                case "aiSession": {
+                    const [row] = await logsDb
+                        .select({ maxId: max(aiSessionLog.id) })
+                        .from(aiSessionLog)
+                        .where(eq(aiSessionLog.orgId, orgId));
+                    return row?.maxId ?? 0;
+                }
             }
         } catch (err) {
             logger.warn(
@@ -670,6 +679,21 @@ export class LogStreamingManager {
                     .limit(limit)) as Array<
                     Record<string, unknown> & { id: number }
                 >;
+
+            case "aiSession":
+                return (await logsDb
+                    .select()
+                    .from(aiSessionLog)
+                    .where(
+                        and(
+                            eq(aiSessionLog.orgId, orgId),
+                            gt(aiSessionLog.id, afterId)
+                        )
+                    )
+                    .orderBy(aiSessionLog.id)
+                    .limit(limit)) as Array<
+                    Record<string, unknown> & { id: number }
+                >;
         }
     }
 
@@ -694,6 +718,14 @@ export class LogStreamingManager {
                 timestamp =
                     typeof row.startedAt === "number" ? row.startedAt : 0;
                 break;
+            case "aiSession":
+                // createdAt is stored as epoch milliseconds; normalise to
+                // epoch seconds to match the other log types.
+                timestamp =
+                    typeof row.createdAt === "number"
+                        ? Math.floor(row.createdAt / 1000)
+                        : 0;
+                break;
         }
 
         const orgId = typeof row.orgId === "string" ? row.orgId : "";

server/private/lib/logStreaming/types.ts

@@ -15,13 +15,14 @@
 // Log type identifiers
 // ---------------------------------------------------------------------------
 
-export type LogType = "request" | "action" | "access" | "connection";
+export type LogType = "request" | "action" | "access" | "connection" | "aiSession";
 
 export const LOG_TYPES: LogType[] = [
     "request",
     "action",
     "access",
-    "connection"
+    "connection",
+    "aiSession"
 ];
 
 // ---------------------------------------------------------------------------

server/private/routers/eventStreamingDestination/createEventStreamingDestination.ts

@@ -37,7 +37,8 @@ const bodySchema = z.strictObject({
     sendConnectionLogs: z.boolean().optional().default(false),
     sendRequestLogs: z.boolean().optional().default(false),
     sendActionLogs: z.boolean().optional().default(false),
-    sendAccessLogs: z.boolean().optional().default(false)
+    sendAccessLogs: z.boolean().optional().default(false),
+    sendAISessionLogs: z.boolean().optional().default(false)
 });
 
 export type CreateEventStreamingDestinationResponse = {
@@ -122,7 +123,8 @@ export async function createEventStreamingDestination(
                 sendAccessLogs: parsedBody.data.sendAccessLogs,
                 sendActionLogs: parsedBody.data.sendActionLogs,
                 sendConnectionLogs: parsedBody.data.sendConnectionLogs,
-                sendRequestLogs: parsedBody.data.sendRequestLogs
+                sendRequestLogs: parsedBody.data.sendRequestLogs,
+                sendAISessionLogs: parsedBody.data.sendAISessionLogs
             })
             .returning();
 

server/private/routers/eventStreamingDestination/listEventStreamingDestinations.ts

@@ -60,6 +60,7 @@ export type ListEventStreamingDestinationsResponse = {
         sendRequestLogs: boolean;
         sendActionLogs: boolean;
         sendAccessLogs: boolean;
+        sendAISessionLogs: boolean;
     }[];
     pagination: {
         total: number;
@@ -83,7 +84,8 @@ const ListEventStreamingDestinationsResponseDataSchema = z.object({
             sendConnectionLogs: z.boolean(),
             sendRequestLogs: z.boolean(),
             sendActionLogs: z.boolean(),
-            sendAccessLogs: z.boolean()
+            sendAccessLogs: z.boolean(),
+            sendAISessionLogs: z.boolean()
         })
     ),
     pagination: z.object({

server/private/routers/eventStreamingDestination/updateEventStreamingDestination.ts

@@ -40,7 +40,8 @@ const bodySchema = z.strictObject({
     sendConnectionLogs: z.boolean().optional(),
     sendRequestLogs: z.boolean().optional(),
     sendActionLogs: z.boolean().optional(),
-    sendAccessLogs: z.boolean().optional()
+    sendAccessLogs: z.boolean().optional(),
+    sendAISessionLogs: z.boolean().optional()
 });
 
 export type UpdateEventStreamingDestinationResponse = {
@@ -125,7 +126,7 @@ export async function updateEventStreamingDestination(
             );
         }
 
-        const { type, config: configToUpdate, enabled, sendAccessLogs, sendActionLogs, sendConnectionLogs, sendRequestLogs } = parsedBody.data;
+        const { type, config: configToUpdate, enabled, sendAccessLogs, sendActionLogs, sendConnectionLogs, sendRequestLogs, sendAISessionLogs } = parsedBody.data;
 
         const updateData: Record<string, unknown> = {
             updatedAt: Date.now()
@@ -141,6 +142,7 @@ export async function updateEventStreamingDestination(
         if (sendActionLogs !== undefined) updateData.sendActionLogs = sendActionLogs;
         if (sendConnectionLogs !== undefined) updateData.sendConnectionLogs = sendConnectionLogs;
         if (sendRequestLogs !== undefined) updateData.sendRequestLogs = sendRequestLogs;
+        if (sendAISessionLogs !== undefined) updateData.sendAISessionLogs = sendAISessionLogs;
 
         await db
             .update(eventStreamingDestinations)

server/private/routers/remoteExitNode/createRemoteExitNode.ts

@@ -191,13 +191,20 @@ export async function createRemoteExitNode(
 
         // If this remote exit node isn't already backing an exit node in
         // another org, we're about to create a brand new one. Reserve a
-        // subnet for it up front so the allocation lock is held across the
-        // whole insert - this guarantees exit node subnets never overlap,
-        // even under concurrent creation, which matters for HA setups.
+        // subnet for it up front, scoped to this org's existing exit nodes,
+        // so the allocation lock is held across the whole insert - this
+        // guarantees exit node subnets never overlap within the org, even
+        // under concurrent creation, which matters for HA setups. Subnets
+        // may still be reused across different orgs; there isn't enough
+        // address space to avoid that, and it isn't necessary since HA only
+        // routes multiple exit nodes for the same org.
         let releaseSubnetLock: (() => Promise<void>) | null = null;
         let newExitNodeAddress: string | null = null;
         if (!existingExitNode) {
-            const { value, release } = await getNextAvailableSubnet();
+            const { value, release } = await getNextAvailableSubnet(
+                db,
+                orgId
+            );
             newExitNodeAddress = value;
             releaseSubnetLock = release;
         }

server/routers/certificates/createCertificate.ts

@@ -23,6 +23,12 @@ export async function createCertificate(
         throw new Error(`Domain with ID ${domainId} not found`);
     }
 
+    // Note: certificates.domain has a global UNIQUE constraint (it is not
+    // scoped per-domainId), so existence must be checked by domain value
+    // alone. Filtering on domainId here as well can cause this check to
+    // miss an existing cert (e.g. if it was stored under a different but
+    // still-valid domainId), leading to an INSERT that then fails on the
+    // unique constraint.
     let existing: Certificate[] = [];
     if (domainRecord.type == "ns" || domainRecord.type == "wildcard") {
         const domainLevelDown = domain.split(".").slice(1).join(".");
@@ -32,16 +38,13 @@ export async function createCertificate(
             .select()
             .from(certificates)
             .where(
-                and(
-                    eq(certificates.domainId, domainId),
-                    or(
-                        eq(certificates.domain, domain),
-                        and(
-                            eq(certificates.wildcard, true),
-                            or(
-                                eq(certificates.domain, domainLevelDown),
-                                eq(certificates.domain, wildcardPrefixed)
-                            )
+                or(
+                    eq(certificates.domain, domain),
+                    and(
+                        eq(certificates.wildcard, true),
+                        or(
+                            eq(certificates.domain, domainLevelDown),
+                            eq(certificates.domain, wildcardPrefixed)
                         )
                     )
                 )
@@ -51,12 +54,7 @@ export async function createCertificate(
         existing = await trx
             .select()
             .from(certificates)
-            .where(
-                and(
-                    eq(certificates.domainId, domainId),
-                    eq(certificates.domain, domain) // exact match for non-NS domains
-                )
-            );
+            .where(eq(certificates.domain, domain)); // exact match for non-NS domains
     }
 
     if (existing.length > 0) {
@@ -87,16 +85,22 @@ export async function createCertificate(
         }
     }
 
-    // No cert found, create a new one in pending state
-    await trx.insert(certificates).values({
-        domain: domainToWrite,
-        domainId,
-        wildcard:
-            domainRecord.type == "ns" ||
-            (domainRecord.type == "wildcard" &&
-                domainRecord.preferWildcardCert), // we can only create wildcard certs for NS domains
-        status: "pending",
-        updatedAt: Math.floor(Date.now() / 1000),
-        createdAt: Math.floor(Date.now() / 1000)
-    });
+    // No cert found, create a new one in pending state. onConflictDoNothing
+    // guards against the domain having been inserted concurrently (or under
+    // a different domainId) between the existence check above and this
+    // insert, since certificates.domain is globally unique.
+    await trx
+        .insert(certificates)
+        .values({
+            domain: domainToWrite,
+            domainId,
+            wildcard:
+                domainRecord.type == "ns" ||
+                (domainRecord.type == "wildcard" &&
+                    domainRecord.preferWildcardCert), // we can only create wildcard certs for NS domains
+            status: "pending",
+            updatedAt: Math.floor(Date.now() / 1000),
+            createdAt: Math.floor(Date.now() / 1000)
+        })
+        .onConflictDoNothing();
 }

server/routers/resource/listResourceUsers.ts

@@ -1,7 +1,7 @@
 import { Request, Response, NextFunction } from "express";
 import { z } from "zod";
 import { db } from "@server/db";
-import { idp, userResources, users } from "@server/db"; // Assuming these are the correct tables
+import { idp, resources, userPolicies, userResources, users } from "@server/db"; // Assuming these are the correct tables
 import { eq } from "drizzle-orm";
 import response from "@server/lib/response";
 import HttpCode from "@server/types/HttpCode";
@@ -14,7 +14,23 @@ const listResourceUsersSchema = z.strictObject({
     resourceId: z.coerce.number().int().positive()
 });
 
-async function queryUsers(resourceId: number) {
+async function queryUsers(resourceId: number, policyId: number | null) {
+    if (policyId !== null) {
+        return await db
+            .select({
+                userId: userPolicies.userId,
+                username: users.username,
+                type: users.type,
+                idpName: idp.name,
+                idpId: users.idpId,
+                email: users.email
+            })
+            .from(userPolicies)
+            .innerJoin(users, eq(userPolicies.userId, users.userId))
+            .leftJoin(idp, eq(users.idpId, idp.idpId))
+            .where(eq(userPolicies.resourcePolicyId, policyId));
+    }
+
     return await db
         .select({
             userId: userResources.userId,
@@ -104,7 +120,26 @@ export async function listResourceUsers(
 
         const { resourceId } = parsedParams.data;
 
-        const resourceUsersList = await queryUsers(resourceId);
+        const [resource] = await db
+            .select()
+            .from(resources)
+            .where(eq(resources.resourceId, resourceId))
+            .limit(1);
+
+        if (!resource) {
+            return next(
+                createHttpError(HttpCode.NOT_FOUND, "Resource not found")
+            );
+        }
+
+        const isInlinePolicy =
+            resource.resourcePolicyId === null &&
+            resource.defaultResourcePolicyId !== null;
+
+        const resourceUsersList = await queryUsers(
+            resourceId,
+            isInlinePolicy ? resource.defaultResourcePolicyId! : null
+        );
 
         return response<ListResourceUsersResponse>(res, {
             data: {