Za ekranami: Projektowanie i budowa systemu kampanii reklamowych w Node.js#

CREATE TABLE campaigns (
  id uuid NOT NULL DEFAULT uuid_generate_v4(),
  name text NOT NULL,
  version integer NOT NULL DEFAULT 1,
  start_at bigint NOT NULL,
  expire_at bigint NOT NULL,
  metadata jsonb,
  status campaigns_status_enum NOT NULL DEFAULT 'draft',
  created_by text,
  idempotency_key uuid,
  created_at timestamp NOT NULL DEFAULT now(),
  updated_at timestamp NOT NULL DEFAULT now(),
  PRIMARY KEY (id)
);

CREATE INDEX idx_campaigns_start_at ON campaigns (start_at);

CREATE TABLE devices (
  device_id text NOT NULL,
  group_id text,
  last_seen timestamptz,
  status devices_status_enum NOT NULL DEFAULT 'offline',
  metadata jsonb,
  PRIMARY KEY (device_id)
);

CREATE TABLE campaign_assets (
  id uuid NOT NULL DEFAULT uuid_generate_v4(),
  campaign_id uuid NOT NULL,
  asset_type campaign_assets_asset_type_enum NOT NULL,
  url text NOT NULL,
  checksum text,
  duration_ms integer,
  size_bytes bigint,
  created_at timestamp NOT NULL DEFAULT now(),
  PRIMARY KEY (id),
  FOREIGN KEY (campaign_id) REFERENCES campaigns(id) ON DELETE CASCADE
);

CREATE TABLE delivery_events (
  id uuid NOT NULL DEFAULT uuid_generate_v4(),
  event_id uuid NOT NULL,
  device_id text NOT NULL,
  campaign_id uuid NOT NULL,
  version integer NOT NULL,
  event_type delivery_events_event_type_enum NOT NULL,
  asset_id uuid,
  payload jsonb,
  created_at timestamp NOT NULL DEFAULT now(),
  PRIMARY KEY (id),
  UNIQUE (event_id),
  FOREIGN KEY (device_id) REFERENCES devices(device_id) ON DELETE CASCADE,
  FOREIGN KEY (campaign_id) REFERENCES campaigns(id) ON DELETE CASCADE
);

CREATE INDEX idx_delivery_events_device_id ON delivery_events (device_id);
CREATE INDEX idx_delivery_events_campaign_id ON delivery_events (campaign_id);

CREATE TABLE outbox (
  id uuid NOT NULL DEFAULT uuid_generate_v4(),
  aggregate_type text NOT NULL,
  aggregate_id uuid NOT NULL,
  event_type text NOT NULL,
  payload jsonb NOT NULL,
  processed boolean NOT NULL DEFAULT false,
  created_at timestamp NOT NULL DEFAULT now(),
  processed_at timestamptz,
  locked_by text,
  locked_at timestamptz,
  PRIMARY KEY (id)
);

CREATE INDEX idx_outbox_processed_locked_at_created_at ON outbox (processed, locked_at, created_at);

async function fetchCampaigns({ page, limit }: FetchCampaignsParams) {
  const offset = (page - 1) * limit;
  const data = await apiFetch<CampaignListResponse>(
    `/campaigns?offset=${offset}&limit=${limit}`,
  );
  return {
    ...data,
    page,
    totalPages: Math.ceil(data.total / limit),
  };
}

async function createCampaign(payload: CreateCampaignPayload) {
  return apiFetch<CampaignResponse>("/campaigns", {
    method: "POST",
    body: JSON.stringify(payload),
  });
}

async function cancelCampaign(id: string) {
  return apiFetch<CampaignResponse>(`/campaigns/${id}/cancel`, {
    method: "POST",
  });
}

function CampaignTable() {
  const [page, setPage] = useState(1);
  const { data, isLoading, isError, error, isPlaceholderData } = useCampaigns({
    page,
    limit: PAGE_SIZE,
  });
  const cancelCampaign = useCancelCampaign();

  if (isLoading) {
    return (
      <Card>
        <CardHeader>
          <CardTitle>Campaigns</CardTitle>
        </CardHeader>
        <CardContent>
          <p className="text-sm text-muted-foreground">Loading campaigns...</p>
        </CardContent>
      </Card>
    );
  }

  if (isError) {
    return (
      <Card>
        <CardHeader>
          <CardTitle>Campaigns</CardTitle>
        </CardHeader>
        <CardContent>
          <p className="text-sm text-destructive">
            {error?.message || "Failed to load campaigns"}
          </p>
        </CardContent>
      </Card>
    );
  }

  const campaigns = data?.data ?? [];
  const totalPages = data?.totalPages ?? 1;
  const total = data?.total ?? 0;

  if (campaigns.length === 0 && page === 1) {
    return (
      <Card>
        <CardHeader>
          <CardTitle>Campaigns</CardTitle>
        </CardHeader>
        <CardContent>
          <p className="text-sm text-muted-foreground">
            No campaigns found. Create one below.
          </p>
        </CardContent>
      </Card>
    );
  }

  return (
    <Card>
      <CardHeader>
        <CardTitle>Campaigns</CardTitle>
      </CardHeader>
      <CardContent className="space-y-4">
        <Table>
          <TableHeader>
            <TableRow>
              <TableHead>Name</TableHead>
              <TableHead>Status</TableHead>
              <TableHead>Start At</TableHead>
              <TableHead>Expire At</TableHead>
              <TableHead>Version</TableHead>
              <TableHead>Created At</TableHead>
              <TableHead className="w-[50px]" />
            </TableRow>
          </TableHeader>
          <TableBody>
            {campaigns.map((campaign) => (
              <TableRow key={campaign.id}>
                <TableCell className="font-medium">{campaign.name}</TableCell>
                <TableCell>
                  <Badge
                    variant={STATUS_VARIANT_MAP[campaign.status] ?? "secondary"}
                  >
                    {campaign.status}
                  </Badge>
                </TableCell>
                <TableCell>{formatDate(campaign.startAt)}</TableCell>
                <TableCell>{formatDate(campaign.expireAt)}</TableCell>
                <TableCell>{campaign.version}</TableCell>
                <TableCell>{formatDate(campaign.createdAt)}</TableCell>
                <TableCell>
                  {campaign.status !== "cancelled" && (
                    <Button
                      variant="ghost"
                      size="icon-sm"
                      disabled={cancelCampaign.isPending}
                      onClick={() => cancelCampaign.mutate(campaign.id)}
                    >
                      <CircleX className="text-muted-foreground" />
                    </Button>
                  )}
                </TableCell>
              </TableRow>
            ))}
          </TableBody>
        </Table>
        <div className="flex items-center justify-between">
          <p className="text-sm text-muted-foreground">
            {total} {total === 1 ? "campaign" : "campaigns"} total
          </p>
          <div className="flex items-center gap-2">
            <Button
              variant="outline"
              size="icon-sm"
              disabled={page <= 1}
              onClick={() => setPage((p) => Math.max(1, p - 1))}
            >
              <ChevronLeft />
            </Button>
            <span className="text-sm tabular-nums">
              {page} / {totalPages}
            </span>
            <Button
              variant="outline"
              size="icon-sm"
              disabled={page >= totalPages || isPlaceholderData}
              onClick={() => setPage((p) => p + 1)}
            >
              <ChevronRight />
            </Button>
          </div>
        </div>
      </CardContent>
    </Card>
  );
}

function CampaignForm() {
  const createCampaign = useCreateCampaign();

  const {
    register,
    handleSubmit,
    control,
    reset,
    formState: { errors },
  } = useForm<CreateCampaignInput>({
    resolver: zodResolver(createCampaignSchema),
    defaultValues: {
      name: "",
      startAt: "",
      expireAt: "",
      assets: [{ assetType: "image", url: "", durationMs: undefined }],
    },
  });

  const { fields, append, remove } = useFieldArray({
    control,
    name: "assets",
  });

  function onSubmit(data: CreateCampaignInput) {
    const idempotencyKey = crypto.randomUUID();

    createCampaign.mutate(
      {
        name: data.name,
        startAt: new Date(data.startAt).getTime(),
        expireAt: new Date(data.expireAt).getTime(),
        assets: data.assets.map((asset) => ({
          assetType: asset.assetType,
          url: asset.url,
          durationMs: asset.durationMs,
        })),
        idempotencyKey,
      },
      {
        onSuccess: () => {
          toast.success("Campaign created successfully");
          reset();
        },
        onError: (error) => {
          toast.error(error.message || "Failed to create campaign");
        },
      },
    );
  }

  return (
    <Card>
      <CardHeader>
        <CardTitle>Create Campaign</CardTitle>
      </CardHeader>
      <CardContent>
        <form onSubmit={handleSubmit(onSubmit)} className="space-y-4">
          <div className="space-y-1.5">
            <Label htmlFor="name">Name</Label>
            <Input
              id="name"
              placeholder="Campaign name"
              {...register("name")}
            />
            {errors.name && (
              <p className="text-sm text-destructive">{errors.name.message}</p>
            )}
          </div>

          <div className="grid grid-cols-2 gap-4">
            <div className="space-y-1.5">
              <Label htmlFor="startAt">Start At</Label>
              <Input
                id="startAt"
                type="datetime-local"
                {...register("startAt")}
              />
              {errors.startAt && (
                <p className="text-sm text-destructive">
                  {errors.startAt.message}
                </p>
              )}
            </div>
            <div className="space-y-1.5">
              <Label htmlFor="expireAt">Expire At</Label>
              <Input
                id="expireAt"
                type="datetime-local"
                {...register("expireAt")}
              />
              {errors.expireAt && (
                <p className="text-sm text-destructive">
                  {errors.expireAt.message}
                </p>
              )}
            </div>
          </div>

          <div className="space-y-2">
            <div className="flex items-center justify-between">
              <Label>Assets</Label>
              <Button
                type="button"
                variant="outline"
                size="sm"
                onClick={() =>
                  append({ assetType: "image", url: "", durationMs: undefined })
                }
              >
                <PlusIcon className="size-4 mr-1" />
                Add Asset
              </Button>
            </div>
            {errors.assets?.root && (
              <p className="text-sm text-destructive">
                {errors.assets.root.message}
              </p>
            )}
            {fields.map((field, index) => (
              <div key={field.id} className="flex items-start gap-2">
                <select
                  {...register(`assets.${index}.assetType`)}
                  className="focus-visible:ring-3 h-8 rounded-lg border border-input bg-transparent px-2.5 text-sm outline-none focus-visible:border-ring focus-visible:ring-ring/50"
                >
                  <option value="image">Image</option>
                  <option value="video">Video</option>
                  <option value="html">HTML</option>
                </select>
                <div className="min-w-0 flex-1">
                  <Input
                    placeholder="Asset URL"
                    {...register(`assets.${index}.url`)}
                  />
                  {errors.assets?.[index]?.url && (
                    <p className="text-sm text-destructive">
                      {errors.assets[index].url.message}
                    </p>
                  )}
                </div>
                <Input
                  type="number"
                  placeholder="Duration (ms)"
                  className="w-36"
                  {...register(`assets.${index}.durationMs`, {
                    setValueAs: (v: string) =>
                      v === "" ? undefined : Number(v),
                  })}
                />
                {fields.length > 1 && (
                  <Button
                    type="button"
                    variant="ghost"
                    size="icon"
                    onClick={() => remove(index)}
                  >
                    <Trash2Icon className="size-4" />
                  </Button>
                )}
              </div>
            ))}
          </div>

          <Button type="submit" disabled={createCampaign.isPending}>
            {createCampaign.isPending ? "Creating..." : "Create Campaign"}
          </Button>
        </form>
      </CardContent>
    </Card>
  );
}

import { Injectable, NotFoundException } from "@nestjs/common";
import { InjectRepository } from "@nestjs/typeorm";
import { DataSource, Repository } from "typeorm";
import { CampaignEntity, CampaignStatusEnum } from "./campaign.entity";
import { CampaignAssetEntity } from "./campaign-asset.entity";
import { OutboxEntity } from "../outbox/outbox.entity";
import { CreateCampaignDto } from "./dtos/create-campaign.dto";
import { CampaignListQueryDto } from "./dtos/campaign-list-response.dto";

@Injectable()
export class CampaignsService {
  constructor(
    @InjectRepository(CampaignEntity)
    private readonly campaignRepository: Repository<CampaignEntity>,
    private readonly dataSource: DataSource,
  ) {}

  async create(dto: CreateCampaignDto) {
    const existing = await this.campaignRepository.findOne({
      where: { idempotencyKey: dto.idempotencyKey },
      relations: ["assets"],
    });

    if (existing) {
      return existing;
    }

    const campaign = await this.dataSource.transaction(async (manager) => {
      const campaignEntity = manager.create(CampaignEntity, {
        name: dto.name,
        startAt: dto.startAt,
        expireAt: dto.expireAt,
        metadata: dto.metadata ?? null,
        idempotencyKey: dto.idempotencyKey,
        status: CampaignStatusEnum.DRAFT,
      });

      campaignEntity.assets = dto.assets.map((asset) =>
        manager.create(CampaignAssetEntity, {
          assetType: asset.assetType,
          url: asset.url,
          durationMs: asset.durationMs ?? null,
        }),
      );

      const savedCampaign = await manager.save(CampaignEntity, campaignEntity);

      const outboxEntry = manager.create(OutboxEntity, {
        aggregateType: "campaign",
        aggregateId: savedCampaign.id,
        eventType: "campaign_created",
        payload: {
          campaignId: savedCampaign.id,
          version: savedCampaign.version,
        },
      });

      await manager.save(OutboxEntity, outboxEntry);

      return savedCampaign;
    });

    return this.campaignRepository.findOne({
      where: { id: campaign.id },
      relations: ["assets"],
    });
  }

  async findAll(query: CampaignListQueryDto) {
    const { status, offset, limit } = query;

    const [data, total] = await this.campaignRepository.findAndCount({
      where: status ? { status } : {},
      relations: ["assets"],
      order: { createdAt: "DESC" },
      skip: offset,
      take: limit,
    });

    return { data, total, offset, limit };
  }

  async findById(id: string) {
    const campaign = await this.campaignRepository.findOne({
      where: { id },
      relations: ["assets"],
    });

    if (!campaign) {
      throw new NotFoundException(`Campaign with id ${id} not found`);
    }

    return campaign;
  }

  async cancel(id: string) {
    const campaign = await this.campaignRepository.findOne({
      where: { id },
    });

    if (!campaign) {
      throw new NotFoundException(`Campaign with id ${id} not found`);
    }

    await this.dataSource.transaction(async (manager) => {
      campaign.status = CampaignStatusEnum.CANCELLED;
      await manager.save(CampaignEntity, campaign);

      const outboxEntry = manager.create(OutboxEntity, {
        aggregateType: "campaign",
        aggregateId: campaign.id,
        eventType: "campaign_cancelled",
        payload: {
          campaignId: campaign.id,
          version: campaign.version,
        },
      });

      await manager.save(OutboxEntity, outboxEntry);
    });

    return this.campaignRepository.findOne({
      where: { id },
      relations: ["assets"],
    });
  }
}

import { Inject, Injectable, Logger, OnModuleDestroy } from "@nestjs/common";
import { DataSource } from "typeorm";
import { InjectQueue } from "@nestjs/bullmq";
import { Queue } from "bullmq";
import { Cron, CronExpression } from "@nestjs/schedule";
import type Redis from "ioredis";
import type Redlock from "redlock";
import { v4 as uuidv4 } from "uuid";
import { OUTBOX_REDIS_CLIENT, OUTBOX_REDLOCK } from "./redlock.provider";

type OutboxRow = {
  id: string;
  event_type: string;
  payload: Record<string, unknown>;
};

const BATCH_SIZE = 10;
const LOCK_TTL_MS = 4000;
const LOCK_KEY = "locks:outbox-poll";
const CLAIM_TTL_MINUTES = 10;

@Injectable()
export class OutboxService implements OnModuleDestroy {
  private readonly logger = new Logger(OutboxService.name);

  constructor(
    @InjectQueue("template-build")
    private readonly templateBuildQueue: Queue,
    @InjectQueue("publish")
    private readonly publishQueue: Queue,
    @Inject(OUTBOX_REDIS_CLIENT)
    private readonly redisClient: Redis,
    @Inject(OUTBOX_REDLOCK)
    private readonly redlock: Redlock,
    private readonly dataSource: DataSource,
  ) {}

  async onModuleDestroy() {
    this.redisClient.disconnect();
  }

  @Cron(CronExpression.EVERY_5_SECONDS)
  async handleCron() {
    let lock;

    try {
      lock = await this.redlock.acquire([LOCK_KEY], LOCK_TTL_MS);
    } catch {
      return;
    }

    try {
      await this.processBatch();
      await this.cleanupStaleClaims();
    } finally {
      await lock.release();
    }
  }

  private async processBatch() {
    const workerId = uuidv4();

    const entries: OutboxRow[] = await this.dataSource.transaction(
      async (manager) => {
        const rows: OutboxRow[] = await manager.query(
          `
        SELECT * FROM outbox
        WHERE processed = false
          AND ("locked_by" IS NULL OR "locked_at" < now() - INTERVAL '${CLAIM_TTL_MINUTES} minutes')
        ORDER BY "created_at" ASC
        LIMIT $1
        FOR UPDATE SKIP LOCKED
        `,
          [BATCH_SIZE],
        );

        if (rows.length === 0) return [];

        const ids = rows.map((r) => r.id);
        await manager.query(
          `UPDATE outbox SET "locked_by" = $1, "locked_at" = now() WHERE id = ANY($2::uuid[]) AND processed = false`,
          [workerId, ids],
        );

        return rows;
      },
    );

    if (entries.length === 0) return;

    let processedCount = 0;

    for (const entry of entries) {
      try {
        const jobOpts = { jobId: `outbox-${entry.id}` };

        switch (entry.event_type) {
          case "campaign_created":
            await this.templateBuildQueue.add(
              entry.event_type,
              entry.payload,
              jobOpts,
            );
            break;
          case "template_ready":
            await this.publishQueue.add(
              entry.event_type,
              entry.payload,
              jobOpts,
            );
            break;
          case "campaign_cancelled":
            await this.publishQueue.add(
              entry.event_type,
              entry.payload,
              jobOpts,
            );
            break;
          default:
            this.logger.warn(
              `Unknown event type: ${entry.event_type} for outbox entry ${entry.id}`,
            );
            break;
        }

        await this.dataSource.query(
          `UPDATE outbox SET processed = true, "processed_at" = now(), "locked_by" = NULL, "locked_at" = NULL WHERE id = $1 AND "locked_by" = $2`,
          [entry.id, workerId],
        );

        processedCount++;
      } catch (error) {
        this.logger.error(
          `Failed to publish outbox entry ${entry.id}: ${error}`,
        );

        try {
          await this.dataSource.query(
            `UPDATE outbox SET "locked_by" = NULL, "locked_at" = NULL WHERE id = $1 AND "locked_by" = $2`,
            [entry.id, workerId],
          );
        } catch (e) {
          this.logger.error(
            `Failed to clear lock for outbox ${entry.id}: ${e}`,
          );
        }
      }
    }

    this.logger.log(
      `Processed ${processedCount}/${entries.length} outbox entries`,
    );
  }

  private async cleanupStaleClaims() {
    try {
      await this.dataSource.query(
        `UPDATE outbox SET "locked_by" = NULL, "locked_at" = NULL WHERE processed = false AND "locked_at" < now() - INTERVAL '${CLAIM_TTL_MINUTES} minutes'`,
      );
    } catch (error) {
      this.logger.error(`Failed to cleanup stale outbox claims: ${error}`);
    }
  }
}

import { Injectable, Logger } from "@nestjs/common";
import { Processor, WorkerHost } from "@nestjs/bullmq";
import { Job } from "bullmq";
import { DataSource } from "typeorm";
import { mkdir, writeFile } from "node:fs/promises";
import { join } from "node:path";
import {
  CampaignEntity,
  CampaignStatusEnum,
} from "../campaigns/campaign.entity";
import { OutboxEntity } from "../outbox/outbox.entity";
import { generateTemplate } from "./template-generator";

@Injectable()
@Processor("template-build")
export class TemplateProcessor extends WorkerHost {
  private readonly logger = new Logger(TemplateProcessor.name);

  constructor(private readonly dataSource: DataSource) {
    super();
  }

  async process(job: Job) {
    const { campaignId } = job.data;

    const campaign = await this.dataSource
      .getRepository(CampaignEntity)
      .findOne({
        where: { id: campaignId },
        relations: ["assets"],
      });

    if (!campaign) {
      throw new Error(`Campaign ${campaignId} not found`);
    }

    await this.dataSource.getRepository(CampaignEntity).update(campaignId, {
      status: CampaignStatusEnum.BUILDING,
    });

    const html = generateTemplate({
      campaignId: campaign.id,
      name: campaign.name,
      startAt: Number(campaign.startAt),
      expireAt: Number(campaign.expireAt),
      assets: campaign.assets.map((a) => ({
        id: a.id,
        assetType: a.assetType,
        url: a.url,
        durationMs: a.durationMs,
      })),
    });

    const dir = join(process.cwd(), "storage", campaignId);
    await mkdir(dir, { recursive: true });
    await writeFile(join(dir, "index.html"), html);

    await this.dataSource.transaction(async (manager) => {
      const campaignToUpdate = await manager.findOne(CampaignEntity, {
        where: { id: campaignId },
      });

      if (!campaignToUpdate) {
        throw new Error(`Campaign ${campaignId} not found`);
      }

      campaignToUpdate.status = CampaignStatusEnum.READY;
      await manager.save(CampaignEntity, campaignToUpdate);

      const outboxEntry = manager.create(OutboxEntity, {
        aggregateType: "campaign",
        aggregateId: campaignId,
        eventType: "template_ready",
        payload: { campaignId, version: campaignToUpdate.version },
      });

      await manager.save(OutboxEntity, outboxEntry);
    });

    this.logger.log(
      `Template built for campaign ${campaignId} v${campaign.version}`,
    );
  }
}

import { Injectable, Logger } from "@nestjs/common";
import { InjectQueue, Processor, WorkerHost } from "@nestjs/bullmq";
import { Job, Queue } from "bullmq";
import { DataSource } from "typeorm";
import { ConfigService } from "@nestjs/config";
import { TOPICS } from "@campaign-system/shared";
import {
  CampaignEntity,
  CampaignStatusEnum,
} from "../campaigns/campaign.entity";
import { MqttProvider } from "./mqtt.provider";
import { DeviceStreamService } from "./device-stream.service";
import { DevicePublishTracker } from "./device-publish-tracker";
import { buildManifest } from "./build-manifest";

const BATCH_SIZE = 200;

@Injectable()
@Processor("publish")
export class PublisherProcessor extends WorkerHost {
  private readonly logger = new Logger(PublisherProcessor.name);

  constructor(
    private readonly dataSource: DataSource,
    private readonly configService: ConfigService,
    private readonly mqttProvider: MqttProvider,
    private readonly deviceStreamService: DeviceStreamService,
    private readonly devicePublishTracker: DevicePublishTracker,
    @InjectQueue("verify-delivery")
    private readonly verifyDeliveryQueue: Queue,
  ) {
    super();
  }

  async process(job: Job) {
    if (job.name === "campaign_cancelled") {
      return this.handleCancel(job);
    }

    return this.handlePublish(job);
  }

  private async handlePublish(job: Job) {
    const { campaignId } = job.data;

    const campaign = await this.dataSource
      .getRepository(CampaignEntity)
      .findOne({
        where: { id: campaignId },
        relations: ["assets"],
      });

    if (!campaign) {
      throw new Error(`Campaign ${campaignId} not found`);
    }

    const startResult = await this.dataSource
      .getRepository(CampaignEntity)
      .createQueryBuilder()
      .update(CampaignEntity)
      .set({ status: CampaignStatusEnum.PUBLISHING })
      .where("id = :id AND status = :expected", {
        id: campaignId,
        expected: CampaignStatusEnum.READY,
      })
      .execute();

    if (startResult.affected === 0) {
      this.logger.warn(
        `Campaign ${campaignId} is no longer in READY state, skipping publish`,
      );
      return;
    }

    const apiBaseUrl = this.configService.get(
      "API_BASE_URL",
      "http://localhost:3000",
    );
    const manifest = buildManifest({ campaign, apiBaseUrl });
    const payload = JSON.stringify(manifest);

    const alreadyPublished = await this.devicePublishTracker.getPublished({
      campaignId: campaign.id,
      version: campaign.version,
    });

    let successCount = 0;
    let failureCount = 0;

    for await (const batch of this.deviceStreamService.streamBatches({
      batchSize: BATCH_SIZE,
      excludeDeviceIds: alreadyPublished,
    })) {
      const results = await Promise.allSettled(
        batch.map((device) =>
          this.mqttProvider.publishWithTimeout({
            topic: TOPICS.NOTIFICATIONS(device.deviceId),
            payload,
            qos: 1,
          }),
        ),
      );

      const publishedIds: string[] = [];

      results.forEach((result, j) => {
        if (result.status === "fulfilled") {
          successCount++;
          publishedIds.push(batch[j]!.deviceId);
        } else {
          failureCount++;
          this.logger.error(
            `Failed to publish to device ${batch[j]?.deviceId}: ${result.reason}`,
          );
        }
      });

      await this.devicePublishTracker.markPublished({
        campaignId: campaign.id,
        version: campaign.version,
        deviceIds: publishedIds,
      });
    }

    await this.verifyDeliveryQueue.add(
      "verify-delivery",
      { campaignId: campaign.id, version: campaign.version },
      {
        delay: 30000,
        jobId: `verify-${campaign.id}-v${campaign.version}-a1`,
      },
    );

    const finalResult = await this.dataSource
      .getRepository(CampaignEntity)
      .createQueryBuilder()
      .update(CampaignEntity)
      .set({ status: CampaignStatusEnum.ACTIVE })
      .where("id = :id AND status = :expected", {
        id: campaignId,
        expected: CampaignStatusEnum.PUBLISHING,
      })
      .execute();

    if (finalResult.affected === 0) {
      this.logger.warn(
        `Campaign ${campaignId} was modified during publishing (likely cancelled), not marking as ACTIVE`,
      );
    }

    this.logger.log(
      `Campaign ${campaignId} published — success: ${successCount}, failed: ${failureCount}, skipped: ${alreadyPublished.size}`,
    );
  }

  private async handleCancel(job: Job) {
    const { campaignId } = job.data;

    const campaign = await this.dataSource
      .getRepository(CampaignEntity)
      .findOne({ where: { id: campaignId } });

    if (!campaign || campaign.status !== CampaignStatusEnum.CANCELLED) {
      this.logger.warn(
        `Campaign ${campaignId} is not in CANCELLED state, skipping revoke`,
      );
      return;
    }

    const payload = JSON.stringify({ type: "revoke", campaignId });

    const alreadyRevoked = await this.devicePublishTracker.getRevoked({
      campaignId,
    });

    let successCount = 0;
    let failureCount = 0;

    for await (const batch of this.deviceStreamService.streamBatches({
      batchSize: BATCH_SIZE,
      excludeDeviceIds: alreadyRevoked,
    })) {
      const results = await Promise.allSettled(
        batch.map((device) =>
          this.mqttProvider.publishWithTimeout({
            topic: TOPICS.CONTROL(device.deviceId),
            payload,
            qos: 1,
          }),
        ),
      );

      const revokedIds: string[] = [];

      results.forEach((result, j) => {
        if (result.status === "fulfilled") {
          successCount++;
          revokedIds.push(batch[j]!.deviceId);
        } else {
          failureCount++;
          this.logger.error(
            `Failed to send revoke to device ${batch[j]?.deviceId}: ${result.reason}`,
          );
        }
      });

      await this.devicePublishTracker.markRevoked({
        campaignId,
        deviceIds: revokedIds,
      });
    }

    await this.verifyDeliveryQueue.add(
      "verify-revoke",
      { campaignId, attempt: 1 },
      {
        delay: 30000,
        jobId: `verify-revoke-${campaignId}-a1`,
      },
    );

    this.logger.log(
      `Campaign ${campaignId} revoke sent — success: ${successCount}, failed: ${failureCount}, skipped: ${alreadyRevoked.size}`,
    );
  }
}

import { Injectable, OnModuleDestroy } from "@nestjs/common";
import { ConfigService } from "@nestjs/config";
import Redis from "ioredis";

const KEY_TTL_SECONDS = 60 * 60 * 24;

type TrackParams = {
  campaignId: string;
  version: number;
  deviceIds: string[];
};

type CampaignVersionParams = {
  campaignId: string;
  version: number;
};

type RevokeTrackParams = {
  campaignId: string;
  deviceIds: string[];
};

type RevokeQueryParams = {
  campaignId: string;
};

@Injectable()
export class DevicePublishTracker implements OnModuleDestroy {
  private readonly redis: Redis;

  constructor(private readonly configService: ConfigService) {
    this.redis = new Redis({
      host: this.configService.get("REDIS_HOST", "localhost"),
      port: this.configService.getOrThrow<number>("REDIS_PORT"),
      keyPrefix: "pubtrack:",
    });
  }

  onModuleDestroy() {
    this.redis.disconnect();
  }

  private buildKey({ campaignId, version }: CampaignVersionParams) {
    return `${campaignId}:v${version}`;
  }

  async markPublished({ campaignId, version, deviceIds }: TrackParams) {
    if (deviceIds.length === 0) return;

    const key = this.buildKey({ campaignId, version });
    const pipeline = this.redis.pipeline();
    pipeline.sadd(key, ...deviceIds);
    pipeline.expire(key, KEY_TTL_SECONDS);
    await pipeline.exec();
  }

  async getPublished({ campaignId, version }: CampaignVersionParams) {
    const key = this.buildKey({ campaignId, version });
    const members = await this.redis.smembers(key);
    return new Set(members);
  }

  private buildRevokeKey({ campaignId }: RevokeQueryParams) {
    return `revoke:${campaignId}`;
  }

  async markRevoked({ campaignId, deviceIds }: RevokeTrackParams) {
    if (deviceIds.length === 0) return;

    const key = this.buildRevokeKey({ campaignId });
    const pipeline = this.redis.pipeline();
    pipeline.sadd(key, ...deviceIds);
    pipeline.expire(key, KEY_TTL_SECONDS);
    await pipeline.exec();
  }

  public async getRevoked({ campaignId }: RevokeQueryParams) {
    const key = this.buildRevokeKey({ campaignId });
    const members = await this.redis.smembers(key);
    return new Set(members);
  }
}

import { Injectable } from "@nestjs/common";
import { DataSource } from "typeorm";
import { DeviceEntity } from "../devices/device.entity";

type StreamBatchesParams = {
  batchSize: number;
  excludeDeviceIds?: Set<string>;
};

@Injectable()
export class DeviceStreamService {
  constructor(private readonly dataSource: DataSource) {}

  async *streamBatches({ batchSize, excludeDeviceIds }: StreamBatchesParams) {
    let cursor: string | null = null;

    while (true) {
      const qb = this.dataSource
        .getRepository(DeviceEntity)
        .createQueryBuilder("device")
        .orderBy("device.deviceId", "ASC")
        .take(batchSize);

      if (cursor) {
        qb.where("device.deviceId > :cursor", { cursor });
      }

      const rows = await qb.getMany();

      if (rows.length === 0) break;

      cursor = rows[rows.length - 1]!.deviceId;

      if (excludeDeviceIds && excludeDeviceIds.size > 0) {
        const filtered = rows.filter((d) => !excludeDeviceIds.has(d.deviceId));
        if (filtered.length > 0) {
          yield filtered;
        }
      } else {
        yield rows;
      }

      if (rows.length < batchSize) break;
    }
  }
}

import { Injectable, Logger } from "@nestjs/common";
import { InjectQueue, Processor, WorkerHost } from "@nestjs/bullmq";
import { Job, Queue } from "bullmq";
import { DataSource } from "typeorm";
import { ConfigService } from "@nestjs/config";
import { TOPICS } from "@campaign-system/shared";
import { CampaignEntity } from "../campaigns/campaign.entity";
import {
  DeliveryEventEntity,
  DeliveryEventTypeEnum,
} from "../ack-consumer/delivery-event.entity";
import { DeviceStreamService } from "./device-stream.service";
import { DevicePublishTracker } from "./device-publish-tracker";
import { buildManifest } from "./build-manifest";
import { MqttProvider } from "../mqtt/mqtt.provider";

const BATCH_SIZE = 200;
const VERIFY_DELAY_MS = 30_000;
const MAX_ATTEMPTS = 3;

type VerifyJobData = {
  campaignId: string;
  version: number;
  attempt?: number;
};

type VerifyRevokeJobData = {
  campaignId: string;
  attempt?: number;
};

@Injectable()
@Processor("verify-delivery")
export class VerifyDeliveryProcessor extends WorkerHost {
  private readonly logger = new Logger(VerifyDeliveryProcessor.name);

  constructor(
    private readonly dataSource: DataSource,
    private readonly configService: ConfigService,
    private readonly mqttProvider: MqttProvider,
    private readonly deviceStreamService: DeviceStreamService,
    private readonly devicePublishTracker: DevicePublishTracker,
    @InjectQueue("verify-delivery")
    private readonly verifyDeliveryQueue: Queue,
  ) {
    super();
  }

  async process(job: Job) {
    if (job.name === "verify-revoke") {
      return this.handleVerifyRevoke(job as Job<VerifyRevokeJobData>);
    }

    return this.handleVerifyDelivery(job as Job<VerifyJobData>);
  }

  private async handleVerifyDelivery(job: Job<VerifyJobData>) {
    const { campaignId, version, attempt = 1 } = job.data;

    this.logger.log(
      `Verifying delivery for campaign ${campaignId} v${version} (attempt ${attempt})`,
    );

    const campaign = await this.dataSource
      .getRepository(CampaignEntity)
      .findOne({ where: { id: campaignId }, relations: ["assets"] });

    if (!campaign) {
      this.logger.warn(`Campaign ${campaignId} not found`);
      return;
    }

    const ackRows = await this.dataSource
      .getRepository(DeliveryEventEntity)
      .find({
        select: { deviceId: true },
        where: {
          campaignId,
          version,
          eventType: DeliveryEventTypeEnum.INSTALL_ACK,
        },
      });

    const ackedDeviceIds = new Set(ackRows.map((r) => r.deviceId));

    const apiBaseUrl = this.configService.get(
      "API_BASE_URL",
      "http://localhost:3000",
    );
    const manifest = buildManifest({ campaign, apiBaseUrl });
    const payload = JSON.stringify(manifest);

    let missingCount = 0;
    let republishCount = 0;

    for await (const batch of this.deviceStreamService.streamBatches({
      batchSize: BATCH_SIZE,
      excludeDeviceIds: ackedDeviceIds,
    })) {
      missingCount += batch.length;

      const results = await Promise.allSettled(
        batch.map((device) =>
          this.mqttProvider.publishWithTimeout({
            topic: TOPICS.NOTIFICATIONS(device.deviceId),
            payload,
            qos: 1,
            retain: true,
            timeout: 5000,
          }),
        ),
      );

      results.forEach((result, j) => {
        if (result.status === "fulfilled") {
          republishCount++;
        } else {
          this.logger.error(
            `Retry publish failed for device ${batch[j]?.deviceId}: ${result.reason}`,
          );
        }
      });
    }

    if (missingCount === 0) {
      this.logger.log(
        `All devices acknowledged campaign ${campaignId} v${version}`,
      );
      return;
    }

    this.logger.log(
      `Re-published to ${republishCount}/${missingCount} missing devices for campaign ${campaignId}`,
    );

    if (attempt < MAX_ATTEMPTS) {
      try {
        const nextAttempt = attempt + 1;
        await this.verifyDeliveryQueue.add(
          "verify-delivery",
          { campaignId, version, attempt: nextAttempt },
          {
            delay: VERIFY_DELAY_MS * Math.pow(2, attempt - 1),
            jobId: `verify-${campaignId}-v${version}-a${nextAttempt}`,
          },
        );
      } catch (e) {
        this.logger.error(`Failed to schedule next verify attempt: ${e}`);
      }
    } else {
      this.logger.log(
        `Max attempts reached for campaign ${campaignId} v${version}`,
      );
    }
  }

  private async handleVerifyRevoke(job: Job<VerifyRevokeJobData>) {
    const { campaignId, attempt = 1 } = job.data;

    this.logger.log(
      `Verifying revoke delivery for campaign ${campaignId} (attempt ${attempt})`,
    );

    const alreadyRevoked = await this.devicePublishTracker.getRevoked({
      campaignId,
    });

    const payload = JSON.stringify({ type: "revoke", campaignId });

    let missingCount = 0;
    let republishCount = 0;

    for await (const batch of this.deviceStreamService.streamBatches({
      batchSize: BATCH_SIZE,
      excludeDeviceIds: alreadyRevoked,
    })) {
      missingCount += batch.length;

      const results = await Promise.allSettled(
        batch.map((device) =>
          this.mqttProvider.publishWithTimeout({
            topic: TOPICS.CONTROL(device.deviceId),
            payload,
            qos: 1,
            timeout: 5000,
          }),
        ),
      );

      const revokedIds: string[] = [];

      results.forEach((result, j) => {
        if (result.status === "fulfilled") {
          republishCount++;
          revokedIds.push(batch[j]!.deviceId);
        } else {
          this.logger.error(
            `Retry revoke failed for device ${batch[j]?.deviceId}: ${result.reason}`,
          );
        }
      });

      await this.devicePublishTracker.markRevoked({
        campaignId,
        deviceIds: revokedIds,
      });
    }

    if (missingCount === 0) {
      this.logger.log(`All devices received revoke for campaign ${campaignId}`);
      return;
    }

    this.logger.log(
      `Re-sent revoke to ${republishCount}/${missingCount} missing devices for campaign ${campaignId}`,
    );

    if (attempt < MAX_ATTEMPTS) {
      const nextAttempt = attempt + 1;
      await this.verifyDeliveryQueue.add(
        "verify-revoke",
        { campaignId, attempt: nextAttempt },
        {
          delay: VERIFY_DELAY_MS * Math.pow(2, attempt - 1),
          jobId: `verify-revoke-${campaignId}-a${nextAttempt}`,
        },
      );
    } else {
      this.logger.log(
        `Max revoke verify attempts reached for campaign ${campaignId}`,
      );
    }
  }
}