Function sessionConsistency

packages/db/src/session-consistency.ts:166–241

()

 *
 */
export function sessionConsistency() {
  return Prisma.defineExtension((client) =>
    client.$extends({
      name: 'session-consistency',
      query: {
        $allOperations: async ({
          operation,
          model,
          args,
          query,
          // This is a hack to force reads to primary when replica hasn't caught up.
          // The readReplicas extension routes queries to primary when in a transaction,
          // so we set __internalParams.transaction = true to achieve this.
          // @ts-expect-error - __internalParams is not in the types
          __internalParams,
        }) => {
          const sessionId = getAlsSessionId();

          // For write operations with session: cache WAL LSN after write
          if (isWriteOperation(operation)) {
            // logger.info('Prisma operation', {
            //   operation,
            //   args,
            //   model,
            // });

            const result = await query(args);

            if (sessionId) {
              // Get current WAL LSN and cache it for this session
              const lsn = await getCurrentWalLsn(client);
              if (lsn) {
                await cacheWalLsnForSession(sessionId, lsn);
                logger.debug(
                  { sessionId, lsn, operation, model },
                  'Cached WAL LSN after write',
                );
              }
            }

            return result;
          }

          // For now, we just force the read to the primary without checking the replica
          // Since the check probably goes to the primary anyways it will always be true,
          // Not sure how to check LSN on the actual replica that will be used for the read.
          if (
            model !== 'Session' &&
            isReadOperation(operation) &&
            sessionId &&
            (await getCachedWalLsn(sessionId))
          ) {
            const MAX_RETRIES = 3;
            const INITIAL_RETRY_DELAY_MS = 50;

            for (let attempt = 0; attempt < MAX_RETRIES; attempt++) {
              const result = await query(args);