illusory-iotam/packages/logic/domains/device/repository.ts

import { ResultAsync, errAsync, okAsync } from "neverthrow";
import { FlowExecCtx } from "@core/flow.execution.context";
import { Database, and, asc, eq } from "@pkg/db";
import { device } from "@pkg/db/schema";
import { type Err } from "@pkg/result";
import { logger } from "@pkg/logger";
import { traceResultAsync } from "@core/observability";
import { CreateDevice, Device, DeviceStatus, UpdateDevice } from "./data";
import { deviceErrors } from "./errors";

export class DeviceRepository {
    constructor(private db: Database) {}

    list(fctx: FlowExecCtx): ResultAsync<Device[], Err> {
        return traceResultAsync({
            name: "device.list",
            fctx,
            fn: () =>
                ResultAsync.fromPromise(
                    this.db.select().from(device).orderBy(asc(device.createdAt)),
                    (e) =>
                        deviceErrors.listFailed(
                            fctx,
                            e instanceof Error ? e.message : String(e),
                        ),
                ).map((rows) => rows as Device[]),
        });
    }

    getById(fctx: FlowExecCtx, id: number): ResultAsync<Device, Err> {
        return traceResultAsync({
            name: "device.getById",
            fctx,
            fn: () =>
                ResultAsync.fromPromise(
                    this.db.query.device.findFirst({ where: eq(device.id, id) }),
                    (e) =>
                        deviceErrors.dbError(
                            fctx,
                            e instanceof Error ? e.message : String(e),
                        ),
                ).andThen((row) => {
                    if (!row) return errAsync(deviceErrors.deviceNotFound(fctx, id));
                    return okAsync(row as Device);
                }),
        });
    }

    create(fctx: FlowExecCtx, data: CreateDevice): ResultAsync<Device, Err> {
        logger.info("Creating device", { ...fctx, host: data.host });

        return traceResultAsync({
            name: "device.create",
            fctx,
            fn: () =>
                ResultAsync.fromPromise(
                    this.db
                        .insert(device)
                        .values({
                            title: data.title,
                            version: data.version,
                            host: data.host,
                            containerId: data.containerId,
                            wsPort: data.wsPort,
                            status: DeviceStatus.OFFLINE,
                            isActive: data.isActive ?? false,
                            inUse: false,
                            createdAt: new Date(),
                            updatedAt: new Date(),
                        })
                        .returning()
                        .execute(),
                    (e) =>
                        deviceErrors.createFailed(
                            fctx,
                            e instanceof Error ? e.message : String(e),
                        ),
                ).map((rows) => rows[0] as Device),
        });
    }

    update(
        fctx: FlowExecCtx,
        id: number,
        updates: UpdateDevice,
    ): ResultAsync<Device, Err> {
        return traceResultAsync({
            name: "device.update",
            fctx,
            fn: () =>
                this.getById(fctx, id).andThen(() =>
                    ResultAsync.fromPromise(
                        this.db
                            .update(device)
                            .set({ ...updates, updatedAt: new Date() })
                            .where(eq(device.id, id))
                            .returning()
                            .execute(),
                        (e) =>
                            deviceErrors.updateFailed(
                                fctx,
                                e instanceof Error ? e.message : String(e),
                            ),
                    ).andThen((rows) => {
                        if (!rows[0])
                            return errAsync(deviceErrors.deviceNotFound(fctx, id));
                        return okAsync(rows[0] as Device);
                    }),
                ),
        });
    }

    delete(fctx: FlowExecCtx, id: number): ResultAsync<boolean, Err> {
        return traceResultAsync({
            name: "device.delete",
            fctx,
            fn: () =>
                this.getById(fctx, id).andThen(() =>
                    ResultAsync.fromPromise(
                        this.db.delete(device).where(eq(device.id, id)).execute(),
                        (e) =>
                            deviceErrors.deleteFailed(
                                fctx,
                                e instanceof Error ? e.message : String(e),
                            ),
                    ).map(() => true),
                ),
        });
    }

    setStatus(
        fctx: FlowExecCtx,
        id: number,
        status: DeviceStatus,
    ): ResultAsync<Device, Err> {
        return this.update(fctx, id, { status });
    }

    allocateIfAvailable(fctx: FlowExecCtx, id: number): ResultAsync<Device, Err> {
        return traceResultAsync({
            name: "device.allocateIfAvailable",
            fctx,
            fn: () =>
                ResultAsync.fromPromise(
                    this.db
                        .update(device)
                        .set({
                            status: DeviceStatus.BUSY,
                            inUse: true,
                            updatedAt: new Date(),
                        })
                        .where(
                            and(
                                eq(device.id, id),
                                eq(device.status, DeviceStatus.ONLINE),
                                eq(device.inUse, false),
                            ),
                        )
                        .returning()
                        .execute(),
                    (e) =>
                        deviceErrors.updateFailed(
                            fctx,
                            e instanceof Error ? e.message : String(e),
                        ),
                ).andThen((rows) => {
                    if (!rows[0]) {
                        return errAsync(deviceErrors.deviceNotAvailable(fctx, id));
                    }
                    return okAsync(rows[0] as Device);
                }),
        });
    }
}