✅ added spanning methods for insights in logic + logger is fully otel-ified as well 🥳

packages/logic/core/observability.ts

@@ -0,0 +1,80 @@
+import { SpanStatusCode, trace, type Attributes } from "@opentelemetry/api";
+import type { FlowExecCtx } from "./flow.execution.context";
+import { ResultAsync } from "neverthrow";
+
+const tracer = trace.getTracer("@pkg/logic");
+
+type BaseSpanOptions = {
+    name: string;
+    fctx?: FlowExecCtx;
+    attributes?: Attributes;
+};
+
+function spanAttributes(
+    fctx?: FlowExecCtx,
+    attributes?: Attributes,
+): Attributes | undefined {
+    const flowAttrs: Attributes = {};
+    if (fctx?.flowId) flowAttrs["flow.id"] = fctx.flowId;
+    if (fctx?.userId) flowAttrs["flow.user_id"] = fctx.userId;
+    if (fctx?.sessionId) flowAttrs["flow.session_id"] = fctx.sessionId;
+
+    if (!attributes && Object.keys(flowAttrs).length === 0) {
+        return undefined;
+    }
+    return { ...flowAttrs, ...(attributes ?? {}) };
+}
+
+export async function withFlowSpan<T>({
+    name,
+    fctx,
+    attributes,
+    fn,
+}: BaseSpanOptions & {
+    fn: () => Promise<T>;
+}): Promise<T> {
+    return tracer.startActiveSpan(
+        name,
+        { attributes: spanAttributes(fctx, attributes) },
+        async (span) => {
+            try {
+                const result = await fn();
+                span.setStatus({ code: SpanStatusCode.OK });
+                return result;
+            } catch (error) {
+                span.recordException(error as Error);
+                span.setStatus({
+                    code: SpanStatusCode.ERROR,
+                    message:
+                        error instanceof Error ? error.message : String(error),
+                });
+                throw error;
+            } finally {
+                span.end();
+            }
+        },
+    );
+}
+
+export function traceResultAsync<T, E>({
+    name,
+    fctx,
+    attributes,
+    fn,
+}: BaseSpanOptions & {
+    fn: () => ResultAsync<T, E>;
+}): ResultAsync<T, E> {
+    return ResultAsync.fromPromise(
+        withFlowSpan({
+            name,
+            fctx,
+            attributes,
+            fn: async () =>
+                fn().match(
+                    (value) => value,
+                    (error) => Promise.reject(error),
+                ),
+        }),
+        (error) => error as E,
+    );
+}

packages/logic/domains/user/account.repository.ts

@@ -1,4 +1,5 @@
 import { FlowExecCtx } from "@/core/flow.execution.context";
+import { traceResultAsync } from "@core/observability";
 import { ERROR_CODES, type Err } from "@pkg/result";
 import { getError, logger } from "@pkg/logger";
 import { auth } from "../auth/config.base";
@@ -34,6 +35,11 @@ export class AccountRepository {
         fctx: FlowExecCtx,
         userId: string,
     ): ResultAsync<boolean, Err> {
+        return traceResultAsync({
+            name: "logic.user.repository.ensureAccountExists",
+            fctx,
+            attributes: { "app.user.id": userId },
+            fn: () => {
                 logger.info("Checking if account exists for user", {
                     ...fctx,
                     userId,
@@ -50,7 +56,9 @@ export class AccountRepository {
                         });
                         return this.dbError(
                             fctx,
-                    error instanceof Error ? error.message : String(error),
+                            error instanceof Error
+                                ? error.message
+                                : String(error),
                         );
                     },
                 ).andThen((existingAccount) => {
@@ -59,7 +67,9 @@ export class AccountRepository {
                             ...fctx,
                             userId,
                         });
-                return ResultAsync.fromSafePromise(Promise.resolve(true));
+                        return ResultAsync.fromSafePromise(
+                            Promise.resolve(true),
+                        );
                     }
 
                     logger.info(
@@ -71,7 +81,9 @@ export class AccountRepository {
                     );
 
                     return ResultAsync.fromPromise(
-                auth.$context.then((ctx) => ctx.password.hash(nanoid())),
+                        auth.$context.then((ctx) =>
+                            ctx.password.hash(nanoid()),
+                        ),
                         (error) => {
                             logger.error("Failed to hash password", {
                                 ...fctx,
@@ -79,7 +91,9 @@ export class AccountRepository {
                             });
                             return this.dbError(
                                 fctx,
-                        error instanceof Error ? error.message : String(error),
+                                error instanceof Error
+                                    ? error.message
+                                    : String(error),
                             );
                         },
                     ).andThen((password) => {
@@ -111,14 +125,19 @@ export class AccountRepository {
                                 );
                             },
                         ).map(() => {
-                    logger.info("Account created successfully for user", {
+                            logger.info(
+                                "Account created successfully for user",
+                                {
                                     ...fctx,
                                     userId,
-                    });
+                                },
+                            );
                             return false;
                         });
                     });
                 });
+            },
+        });
     }
 
     rotatePassword(
@@ -126,6 +145,11 @@ export class AccountRepository {
         userId: string,
         password: string,
     ): ResultAsync<string, Err> {
+        return traceResultAsync({
+            name: "logic.user.repository.rotatePassword",
+            fctx,
+            attributes: { "app.user.id": userId },
+            fn: () => {
                 logger.info("Starting password rotation for user", {
                     ...fctx,
                     userId,
@@ -138,14 +162,13 @@ export class AccountRepository {
                     (error) => {
                         logger.error(
                             "Failed to check account existence for password rotation",
-                    {
+                            { ...fctx, error },
-                        ...fctx,
-                        error,
-                    },
                         );
                         return this.dbError(
                             fctx,
-                    error instanceof Error ? error.message : String(error),
+                            error instanceof Error
+                                ? error.message
+                                : String(error),
                         );
                     },
                 ).andThen((existingAccount) => {
@@ -162,15 +185,22 @@ export class AccountRepository {
                     }
 
                     return ResultAsync.fromPromise(
-                auth.$context.then((ctx) => ctx.password.hash(password)),
+                        auth.$context.then((ctx) =>
+                            ctx.password.hash(password),
+                        ),
                         (error) => {
-                    logger.error("Failed to hash password for rotation", {
+                            logger.error(
+                                "Failed to hash password for rotation",
+                                {
                                     ...fctx,
                                     error,
-                    });
+                                },
+                            );
                             return this.dbError(
                                 fctx,
-                        error instanceof Error ? error.message : String(error),
+                                error instanceof Error
+                                    ? error.message
+                                    : String(error),
                             );
                         },
                     ).andThen((hashed) => {
@@ -198,9 +228,10 @@ export class AccountRepository {
                                 );
                             },
                         ).map((result) => {
-                    logger.info("User's password updated successfully", {
+                            logger.info(
-                        ...fctx,
+                                "User's password updated successfully",
-                    });
+                                { ...fctx },
+                            );
                             logger.debug("Password rotation result", {
                                 ...fctx,
                                 result,
@@ -209,5 +240,7 @@ export class AccountRepository {
                         });
                     });
                 });
+            },
+        });
     }
 }

packages/logic/domains/user/controller.ts

@@ -1,4 +1,5 @@
 import { FlowExecCtx } from "@/core/flow.execution.context";
+import { traceResultAsync } from "@core/observability";
 import { AccountRepository } from "./account.repository";
 import { UserRepository } from "./repository";
 import { db } from "@pkg/db";
@@ -10,19 +11,39 @@ export class UserController {
     ) {}
 
     getUserInfo(fctx: FlowExecCtx, userId: string) {
-        return this.userRepository.getUserInfo(fctx, userId);
+        return traceResultAsync({
+            name: "logic.user.controller.getUserInfo",
+            fctx,
+            attributes: { "app.user.id": userId },
+            fn: () => this.userRepository.getUserInfo(fctx, userId),
+        });
     }
 
     ensureAccountExists(fctx: FlowExecCtx, userId: string) {
-        return this.accountRepo.ensureAccountExists(fctx, userId);
+        return traceResultAsync({
+            name: "logic.user.controller.ensureAccountExists",
+            fctx,
+            attributes: { "app.user.id": userId },
+            fn: () => this.accountRepo.ensureAccountExists(fctx, userId),
+        });
     }
 
     isUsernameAvailable(fctx: FlowExecCtx, username: string) {
-        return this.userRepository.isUsernameAvailable(fctx, username);
+        return traceResultAsync({
+            name: "logic.user.controller.isUsernameAvailable",
+            fctx,
+            attributes: { "app.user.username": username },
+            fn: () => this.userRepository.isUsernameAvailable(fctx, username),
+        });
     }
 
     updateLastVerified2FaAtToNow(fctx: FlowExecCtx, userId: string) {
-        return this.userRepository.updateLastVerified2FaAtToNow(fctx, userId);
+        return traceResultAsync({
+            name: "logic.user.controller.updateLastVerified2FaAtToNow",
+            fctx,
+            attributes: { "app.user.id": userId },
+            fn: () => this.userRepository.updateLastVerified2FaAtToNow(fctx, userId),
+        });
     }
 
     banUser(
@@ -31,19 +52,39 @@ export class UserController {
         reason: string,
         banExpiresAt: Date,
     ) {
-        return this.userRepository.banUser(fctx, userId, reason, banExpiresAt);
+        return traceResultAsync({
+            name: "logic.user.controller.banUser",
+            fctx,
+            attributes: { "app.user.id": userId },
+            fn: () => this.userRepository.banUser(fctx, userId, reason, banExpiresAt),
+        });
     }
 
     isUserBanned(fctx: FlowExecCtx, userId: string) {
-        return this.userRepository.isUserBanned(fctx, userId);
+        return traceResultAsync({
+            name: "logic.user.controller.isUserBanned",
+            fctx,
+            attributes: { "app.user.id": userId },
+            fn: () => this.userRepository.isUserBanned(fctx, userId),
+        });
     }
 
     getBanInfo(fctx: FlowExecCtx, userId: string) {
-        return this.userRepository.getBanInfo(fctx, userId);
+        return traceResultAsync({
+            name: "logic.user.controller.getBanInfo",
+            fctx,
+            attributes: { "app.user.id": userId },
+            fn: () => this.userRepository.getBanInfo(fctx, userId),
+        });
     }
 
     rotatePassword(fctx: FlowExecCtx, userId: string, password: string) {
-        return this.accountRepo.rotatePassword(fctx, userId, password);
+        return traceResultAsync({
+            name: "logic.user.controller.rotatePassword",
+            fctx,
+            attributes: { "app.user.id": userId },
+            fn: () => this.accountRepo.rotatePassword(fctx, userId, password),
+        });
     }
 }
 

packages/logic/domains/user/repository.ts

@@ -1,5 +1,6 @@
 import { ResultAsync, errAsync, okAsync } from "neverthrow";
 import { FlowExecCtx } from "@core/flow.execution.context";
+import { traceResultAsync } from "@core/observability";
 import { type Err } from "@pkg/result";
 import { Database, eq } from "@pkg/db";
 import { BanInfo, User } from "./data";
@@ -11,6 +12,11 @@ export class UserRepository {
     constructor(private db: Database) {}
 
     getUserInfo(fctx: FlowExecCtx, userId: string): ResultAsync<User, Err> {
+        return traceResultAsync({
+            name: "logic.user.repository.getUserInfo",
+            fctx,
+            attributes: { "app.user.id": userId },
+            fn: () => {
                 logger.info("Getting user info for user", {
                     flowId: fctx.flowId,
                     userId,
@@ -45,38 +51,7 @@ export class UserRepository {
                     });
                     return okAsync(userData as User);
                 });
-    }
-
-    updateLastVerified2FaAtToNow(
-        fctx: FlowExecCtx,
-        userId: string,
-    ): ResultAsync<boolean, Err> {
-        logger.info("Updating last 2FA verified timestamp for user", {
-            flowId: fctx.flowId,
-            userId,
-        });
-
-        return ResultAsync.fromPromise(
-            this.db
-                .update(user)
-                .set({ last2FAVerifiedAt: new Date() })
-                .where(eq(user.id, userId))
-                .execute(),
-            (error) => {
-                logger.error("Failed to update last 2FA verified timestamp", {
-                    ...fctx,
-                    error,
-                });
-                return userErrors.updateFailed(
-                    fctx,
-                    error instanceof Error ? error.message : String(error),
-                );
             },
-        ).map(() => {
-            logger.info("Last 2FA verified timestamp updated successfully", {
-                ...fctx,
-            });
-            return true;
         });
     }
 
@@ -84,6 +59,11 @@ export class UserRepository {
         fctx: FlowExecCtx,
         username: string,
     ): ResultAsync<boolean, Err> {
+        return traceResultAsync({
+            name: "logic.user.repository.isUsernameAvailable",
+            fctx,
+            attributes: { "app.user.username": username },
+            fn: () => {
                 logger.info("Checking username availability", {
                     ...fctx,
                     username,
@@ -112,6 +92,54 @@ export class UserRepository {
                     });
                     return isAvailable;
                 });
+            },
+        });
+    }
+
+    updateLastVerified2FaAtToNow(
+        fctx: FlowExecCtx,
+        userId: string,
+    ): ResultAsync<boolean, Err> {
+        return traceResultAsync({
+            name: "logic.user.repository.updateLastVerified2FaAtToNow",
+            fctx,
+            attributes: { "app.user.id": userId },
+            fn: () => {
+                logger.info("Updating last 2FA verified timestamp for user", {
+                    flowId: fctx.flowId,
+                    userId,
+                });
+
+                return ResultAsync.fromPromise(
+                    this.db
+                        .update(user)
+                        .set({ last2FAVerifiedAt: new Date() })
+                        .where(eq(user.id, userId))
+                        .execute(),
+                    (error) => {
+                        logger.error(
+                            "Failed to update last 2FA verified timestamp",
+                            {
+                                ...fctx,
+                                error,
+                            },
+                        );
+                        return userErrors.updateFailed(
+                            fctx,
+                            error instanceof Error ? error.message : String(error),
+                        );
+                    },
+                ).map(() => {
+                    logger.info(
+                        "Last 2FA verified timestamp updated successfully",
+                        {
+                            ...fctx,
+                        },
+                    );
+                    return true;
+                });
+            },
+        });
     }
 
     banUser(
@@ -120,6 +148,11 @@ export class UserRepository {
         reason: string,
         banExpiresAt: Date,
     ): ResultAsync<boolean, Err> {
+        return traceResultAsync({
+            name: "logic.user.repository.banUser",
+            fctx,
+            attributes: { "app.user.id": userId },
+            fn: () => {
                 logger.info("Banning user", {
                     ...fctx,
                     userId,
@@ -152,9 +185,16 @@ export class UserRepository {
                     });
                     return true;
                 });
+            },
+        });
     }
 
     isUserBanned(fctx: FlowExecCtx, userId: string): ResultAsync<boolean, Err> {
+        return traceResultAsync({
+            name: "logic.user.repository.isUserBanned",
+            fctx,
+            attributes: { "app.user.id": userId },
+            fn: () => {
                 logger.info("Checking ban status for user", { ...fctx, userId });
 
                 return ResultAsync.fromPromise(
@@ -183,13 +223,11 @@ export class UserRepository {
                         return errAsync(userErrors.userNotFound(fctx));
                     }
 
-            // If not banned, return false
                     if (!userData.banned) {
                         logger.info("User is not banned", { ...fctx, userId });
                         return okAsync(false);
                     }
 
-            // If banned but no expiry date, consider permanently banned
                     if (!userData.banExpires) {
                         logger.info("User is permanently banned", { ...fctx, userId });
                         return okAsync(true);
@@ -237,7 +275,6 @@ export class UserRepository {
                                     "Failed to unban user after expiry, still returning banned status",
                                     { ...fctx, userId, error },
                                 );
-                        // Still return banned status since we couldn't update
                                 return okAsync(true);
                             });
                     }
@@ -249,9 +286,16 @@ export class UserRepository {
                     });
                     return okAsync(true);
                 });
+            },
+        });
     }
 
     getBanInfo(fctx: FlowExecCtx, userId: string): ResultAsync<BanInfo, Err> {
+        return traceResultAsync({
+            name: "logic.user.repository.getBanInfo",
+            fctx,
+            attributes: { "app.user.id": userId },
+            fn: () => {
                 logger.info("Getting ban info for user", { ...fctx, userId });
 
                 return ResultAsync.fromPromise(
@@ -285,5 +329,7 @@ export class UserRepository {
                         expires: userData.banExpires || undefined,
                     });
                 });
+            },
+        });
     }
 }

packages/logic/package.json

@@ -5,6 +5,7 @@
     },
     "dependencies": {
         "@hono/standard-validator": "^0.2.1",
+        "@opentelemetry/api": "^1.9.0",
         "@otplib/plugin-base32-scure": "^13.3.0",
         "@otplib/plugin-crypto-noble": "^13.3.0",
         "@otplib/totp": "^13.3.0",

pnpm-lock.yaml

@@ -292,6 +292,9 @@ importers:
       '@hono/standard-validator':
         specifier: ^0.2.1
         version: 0.2.2(@standard-schema/spec@1.1.0)(hono@4.12.3)
+      '@opentelemetry/api':
+        specifier: ^1.9.0
+        version: 1.9.0
       '@otplib/plugin-base32-scure':
         specifier: ^13.3.0
         version: 13.3.0