ThingsBoard Connector Plugin - Ingest & Telemetry Decoupling (Plan 3 of 3) Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Make the HMP platform its own source of truth: ingest telemetry per-organization, serve all telemetry reads and the live monitoring stream from the internal TimescaleDB/Redis pipeline, migrate every remaining ThingsBoard consumer onto the Plan 1 connector, then drop the legacy TB columns and dead MQTT/EMQX cruft.

Architecture: A per-org ingest webhook resolves the org from a per-connection token and reuses the existing internal TelemetryService.save() pipeline (TimescaleDB + Redis + /ws emit + automation). Read methods (getLatest/getHistory/getPondTelemetry/exportTelemetry and monitoring snapshot/history) are rewritten to query device_telemetry. The /monitoring live stream is fed from an internal telemetry event bus emitted by save() instead of the TB WebSocket. Device provisioning and RPC move from the concrete ThingsBoardClientService to the connector (ConnectorConnectionManager + DeviceIntegration mapping). Finally the old TB columns/module/env are removed.

Tech Stack: NestJS 11, TypeORM 0.3 (Postgres/TimescaleDB), Redis, Socket.io, Jest.

Prerequisites: Plans 1 and 2 merged. This plan uses ConnectorConnectionManager, DeviceIntegration, IntegrationConnection, CredentialCipher from @/integrations/....

Design spec: docs/superpowers/specs/2026-07-11-thingsboard-connector-plugin-design.md Sections 4.4, 7, 8, 11.

Critical facts from codebase exploration (do not re-derive):

TelemetryService.save() (backend/src/telemetry/telemetry.service.ts:59-91) is ALREADY fully internal (TimescaleDB write, Redis telemetry:latest:{deviceId} TTL 300s, /ws emitTelemetry, automation). Only the READ methods still call TB.
A working internal-DB history read already exists at backend/src/devices/devices.service.ts:601-669 (getTelemetry) producing the same TelemetryDataPoint[] / aggregated shapes - reuse its query pattern.
device_telemetry is a wide/document table: PK (time, device_id), all metrics in one jsonb data column. It is NOT actually a hypertable (the ConfigureTimescaleDB migration is a no-op) - reads are plain table scans; acceptable at current scale, flagged for follow-up.
The /ws frontend telemetry stream is already internal; only the /monitoring namespace stream still uses the TB WS adapter.
checkThingsBoard currently gates overall platform health (health.service.ts:21).

File Structure

Backend - create:

backend/src/integrations/services/device-integration.service.ts (+ .spec.ts) - resolve a device’s external id / credentials from device_integrations.
backend/src/telemetry/dto/org-ingest.dto.ts - per-org ingest payload (telemetry + attributes).
backend/src/telemetry/guards/org-ingest.guard.ts (+ .spec.ts) - resolves org from x-ingest-token.
backend/src/telemetry/controllers/integration-ingest.controller.ts - new /integrations/thingsboard/ingest route.
backend/src/telemetry/internal-telemetry-bus.ts - in-process telemetry event emitter for the monitoring stream.
backend/src/migrations/1778940000000-BackfillDeviceIntegrationsFromDevices.ts
backend/src/migrations/1778950000000-DropLegacyDeviceTbColumns.ts

Backend - modify:

backend/src/telemetry/telemetry.service.ts - add ingestForOrg, rewrite reads to DB, emit to the internal bus in save, drop ThingsBoardClientService injection.
backend/src/telemetry/telemetry.module.ts - register new controller/guard/bus, import IntegrationsModule, drop ThingsboardModule.
backend/src/monitoring/services/monitoring-query.service.ts - snapshot/history from DB.
backend/src/monitoring/services/monitoring-stream.service.ts - subscribe to internal bus instead of TB adapter.
backend/src/monitoring/services/monitoring-command.service.ts - RPC via connector + DeviceIntegrationService.
backend/src/monitoring/services/monitoring-health.service.ts - internal health signal.
backend/src/monitoring/monitoring.module.ts - wiring.
backend/src/devices/devices.service.ts - provision/rollback/remove via connector + DeviceIntegration.
backend/src/devices/device-command-dispatch.service.ts - RPC via connector.
backend/src/devices/device-activity-monitor.service.ts - offline from lastSeenAt, drop TB poll.
backend/src/devices/system-device-inventory.service.ts - remote delete via connector.
backend/src/devices/devices.module.ts - import IntegrationsModule, drop ThingsboardModule.
backend/src/devices/entities/device.entity.ts - remove mqttTopic, thingsboardDeviceId, accessToken, secret + toJSON.
backend/src/health/health.service.ts - demote TB to non-fatal.
backend/src/app.module.ts - remove ThingsboardModule.
backend/CLAUDE.md, .env.production - remove stale MQTT/EMQX docs and config.

Backend - delete:

backend/src/thingsboard/ (entire module) once no consumer imports it.

Task 1: `DeviceIntegrationService` - resolve external id / credentials

Every consumer that used device.thingsboardDeviceId will instead resolve the external id from device_integrations.

Files:

Create: backend/src/integrations/services/device-integration.service.ts
Test: backend/src/integrations/services/device-integration.service.spec.ts
Modify: backend/src/integrations/integrations.module.ts (provide + export it)
Step 1: Write the failing test

// backend/src/integrations/services/device-integration.service.spec.ts
import type { Repository } from 'typeorm';
import { CredentialCipher } from '@/integrations/crypto/credential-cipher';
import { DeviceIntegration } from '@/integrations/entities/device-integration.entity';
import { DeviceIntegrationService } from '@/integrations/services/device-integration.service';

describe('DeviceIntegrationService', () => {
  const cipher = new CredentialCipher(Buffer.alloc(32, 2).toString('base64'));
  let repo: { findOne: jest.Mock };
  let service: DeviceIntegrationService;

  beforeEach(() => {
    repo = { findOne: jest.fn() };
    service = new DeviceIntegrationService(
      repo as unknown as Repository<DeviceIntegration>,
      cipher,
    );
  });

  it('returns the external id for a mapped device', async () => {
    repo.findOne.mockResolvedValue({ externalId: 'tb-1' } as DeviceIntegration);
    expect(await service.resolveExternalId('dev-1', 'thingsboard')).toBe('tb-1');
  });

  it('returns null when the device is not mapped', async () => {
    repo.findOne.mockResolvedValue(null);
    expect(await service.resolveExternalId('dev-1', 'thingsboard')).toBeNull();
  });

  it('decrypts stored credentials', async () => {
    repo.findOne.mockResolvedValue({
      externalId: 'tb-1',
      credentialsEncrypted: cipher.encrypt('tok'),
    } as DeviceIntegration);
    expect(await service.resolveAccessToken('dev-1', 'thingsboard')).toBe('tok');
  });
});

Step 2: Run test to verify it fails

Run: make backend-test FILE=backend/src/integrations/services/device-integration.service.spec.ts
Expected: FAIL - cannot find module.

Step 3: Write the implementation

// backend/src/integrations/services/device-integration.service.ts
import { Injectable } from '@nestjs/common';
import { InjectRepository } from '@nestjs/typeorm';
import { Repository } from 'typeorm';
import { CredentialCipher } from '@/integrations/crypto/credential-cipher';
import { DeviceIntegration } from '@/integrations/entities/device-integration.entity';

@Injectable()
export class DeviceIntegrationService {
  constructor(
    @InjectRepository(DeviceIntegration)
    private readonly repo: Repository<DeviceIntegration>,
    private readonly cipher: CredentialCipher,
  ) {}

  async resolveExternalId(deviceId: string, provider = 'thingsboard'): Promise<string | null> {
    const mapping = await this.repo.findOne({ where: { deviceId, provider } });
    return mapping?.externalId ?? null;
  }

  async resolveAccessToken(deviceId: string, provider = 'thingsboard'): Promise<string | null> {
    const mapping = await this.repo.findOne({ where: { deviceId, provider } });
    if (!mapping?.credentialsEncrypted) {
      return null;
    }
    return this.cipher.decrypt(mapping.credentialsEncrypted);
  }

  async upsertMapping(input: {
    deviceId: string;
    organizationId: string;
    provider: string;
    externalId: string;
    accessToken?: string | null;
  }): Promise<DeviceIntegration> {
    const existing = await this.repo.findOne({
      where: { deviceId: input.deviceId, provider: input.provider },
    });
    const row: DeviceIntegration = (existing ?? new DeviceIntegration()) as DeviceIntegration;
    row.deviceId = input.deviceId;
    row.organizationId = input.organizationId;
    row.provider = input.provider;
    row.externalId = input.externalId;
    row.credentialsEncrypted = input.accessToken ? this.cipher.encrypt(input.accessToken) : null;
    row.lastSyncedAt = new Date();
    return this.repo.save(row);
  }

  async deleteMapping(deviceId: string, provider = 'thingsboard'): Promise<void> {
    await this.repo.delete({ deviceId, provider });
  }
}

Step 4: Register in the module

In backend/src/integrations/integrations.module.ts, add DeviceIntegrationService to providers and to exports, and import it at the top. (DeviceIntegration is already in forFeature from Plan 1.)

Step 5: Run test to verify it passes

Run: make backend-test FILE=backend/src/integrations/services/device-integration.service.spec.ts
Expected: PASS (3 tests).

Step 6: Commit

git add backend/src/integrations/services/device-integration.service.ts backend/src/integrations/services/device-integration.service.spec.ts backend/src/integrations/integrations.module.ts
git commit -m "feat(integrations): add DeviceIntegrationService for external-id resolution"

Task 2: Per-org ingest DTO + guard

The guard resolves the org from x-ingest-token against integration_connections.ingest_token, and attaches the connection to the request.

Files:

Create: backend/src/telemetry/dto/org-ingest.dto.ts
Create: backend/src/telemetry/guards/org-ingest.guard.ts
Test: backend/src/telemetry/guards/org-ingest.guard.spec.ts
Step 1: Write the DTO

Extends the existing telemetry payload with an optional attributes map (device metadata sync).

// backend/src/telemetry/dto/org-ingest.dto.ts
import { IsNotEmpty, IsNumber, IsObject, IsOptional, IsString } from 'class-validator';

export class OrgIngestDto {
  @IsString()
  @IsNotEmpty()
  deviceName: string; // == HMP serialNumber

  @IsOptional()
  @IsNumber()
  ts?: number; // epoch ms

  @IsOptional()
  @IsObject()
  values?: Record<string, number | string | boolean>;

  @IsOptional()
  @IsObject()
  attributes?: Record<string, number | string | boolean>;
}

Step 2: Write the failing guard test

// backend/src/telemetry/guards/org-ingest.guard.spec.ts
import { UnauthorizedException } from '@nestjs/common';
import type { ExecutionContext } from '@nestjs/common';
import type { Repository } from 'typeorm';
import { IntegrationConnection } from '@/integrations/entities/integration-connection.entity';
import { OrgIngestGuard } from '@/telemetry/guards/org-ingest.guard';

function ctxWithToken(token?: string): { ctx: ExecutionContext; req: Record<string, unknown> } {
  const req: Record<string, unknown> = { headers: token ? { 'x-ingest-token': token } : {} };
  const ctx = {
    switchToHttp: () => ({ getRequest: () => req }),
  } as unknown as ExecutionContext;
  return { ctx, req };
}

describe('OrgIngestGuard', () => {
  let repo: { findOne: jest.Mock };
  let guard: OrgIngestGuard;

  beforeEach(() => {
    repo = { findOne: jest.fn() };
    guard = new OrgIngestGuard(repo as unknown as Repository<IntegrationConnection>);
  });

  it('rejects a missing token', async () => {
    const { ctx } = ctxWithToken(undefined);
    await expect(guard.canActivate(ctx)).rejects.toBeInstanceOf(UnauthorizedException);
  });

  it('rejects an unknown token', async () => {
    repo.findOne.mockResolvedValue(null);
    const { ctx } = ctxWithToken('nope');
    await expect(guard.canActivate(ctx)).rejects.toBeInstanceOf(UnauthorizedException);
  });

  it('accepts a valid token and attaches the connection to the request', async () => {
    const connection = { id: 'c1', organizationId: 'org-1' } as IntegrationConnection;
    repo.findOne.mockResolvedValue(connection);
    const { ctx, req } = ctxWithToken('valid');
    await expect(guard.canActivate(ctx)).resolves.toBe(true);
    expect(req.integrationConnection).toBe(connection);
  });
});

Step 3: Run test to verify it fails

Run: make backend-test FILE=backend/src/telemetry/guards/org-ingest.guard.spec.ts
Expected: FAIL - cannot find module.

Step 4: Write the guard

// backend/src/telemetry/guards/org-ingest.guard.ts
import { CanActivate, ExecutionContext, Injectable, UnauthorizedException } from '@nestjs/common';
import { InjectRepository } from '@nestjs/typeorm';
import { Repository } from 'typeorm';
import { IntegrationConnection } from '@/integrations/entities/integration-connection.entity';

@Injectable()
export class OrgIngestGuard implements CanActivate {
  constructor(
    @InjectRepository(IntegrationConnection)
    private readonly connectionRepo: Repository<IntegrationConnection>,
  ) {}

  async canActivate(context: ExecutionContext): Promise<boolean> {
    const request = context.switchToHttp().getRequest<{
      headers: Record<string, string | string[] | undefined>;
      integrationConnection?: IntegrationConnection;
    }>();
    const token = request.headers['x-ingest-token'];
    if (!token || typeof token !== 'string') {
      throw new UnauthorizedException('Missing ingest token');
    }
    const connection = await this.connectionRepo.findOne({ where: { ingestToken: token } });
    if (!connection) {
      throw new UnauthorizedException('Invalid ingest token');
    }
    request.integrationConnection = connection;
    return true;
  }
}

Step 5: Run test to verify it passes

Run: make backend-test FILE=backend/src/telemetry/guards/org-ingest.guard.spec.ts
Expected: PASS (3 tests).

Step 6: Commit

git add backend/src/telemetry/dto/org-ingest.dto.ts backend/src/telemetry/guards/org-ingest.guard.ts backend/src/telemetry/guards/org-ingest.guard.spec.ts
git commit -m "feat(telemetry): add per-org ingest DTO and token guard"

Task 3: `TelemetryService.ingestForOrg` + attribute sync

Org-scoped ingest: match device by { serialNumber, organizationId }, write telemetry via the existing save(), and sync attributes onto the device.

Files:

Modify: backend/src/telemetry/telemetry.service.ts
Test: backend/src/telemetry/telemetry.service.spec.ts (add cases; create if absent)
Step 1: Write the failing test

Add to the telemetry service spec (mock deviceRepository + save):

// in backend/src/telemetry/telemetry.service.spec.ts
describe('ingestForOrg', () => {
  it('matches the device within the org and saves telemetry', async () => {
    deviceRepository.findOne.mockResolvedValue({ id: 'dev-1', serialNumber: 'SN-1' });
    const saveSpy = jest.spyOn(service, 'save').mockResolvedValue(undefined);

    const accepted = await service.ingestForOrg('org-1', {
      deviceName: 'SN-1',
      ts: 1000,
      values: { ph: 7 },
    });

    expect(deviceRepository.findOne).toHaveBeenCalledWith({
      where: { serialNumber: 'SN-1', organizationId: 'org-1' },
    });
    expect(saveSpy).toHaveBeenCalledWith(
      expect.objectContaining({ deviceId: 'dev-1', data: { ph: 7 } }),
    );
    expect(accepted).toBe(true);
  });

  it('ignores an unknown device in that org', async () => {
    deviceRepository.findOne.mockResolvedValue(null);
    const accepted = await service.ingestForOrg('org-1', { deviceName: 'SN-X', values: { ph: 7 } });
    expect(accepted).toBe(false);
  });

  it('syncs attributes onto the device without requiring values', async () => {
    deviceRepository.findOne.mockResolvedValue({ id: 'dev-1', serialNumber: 'SN-1' });
    jest.spyOn(service, 'save').mockResolvedValue(undefined);

    await service.ingestForOrg('org-1', { deviceName: 'SN-1', attributes: { firmware: '1.2.3' } });

    expect(deviceRepository.update).toHaveBeenCalledWith(
      'dev-1',
      expect.objectContaining({ capabilities: expect.objectContaining({ firmware: '1.2.3' }) }),
    );
  });
});

Step 2: Run test to verify it fails

Run: make backend-test FILE=backend/src/telemetry/telemetry.service.spec.ts
Expected: FAIL - ingestForOrg not a function.

Step 3: Add the method

Add to TelemetryService (near the existing ingestFromThingsBoard, telemetry.service.ts:100):

  async ingestForOrg(organizationId: string, dto: OrgIngestDto): Promise<boolean> {
    const device = await this.deviceRepository.findOne({
      where: { serialNumber: dto.deviceName, organizationId },
    });
    if (!device) {
      this.logger.warn(
        `Ignoring ingest for unknown device ${dto.deviceName} in org ${organizationId}`,
      );
      return false;
    }

    const timestamp = dto.ts ? new Date(dto.ts) : new Date();
    await this.deviceRepository.update(device.id, { status: 'online', lastSeenAt: timestamp });

    if (dto.attributes && Object.keys(dto.attributes).length > 0) {
      await this.deviceRepository.update(device.id, {
        capabilities: { ...(device.capabilities ?? {}), ...dto.attributes },
      });
    }

    if (dto.values && Object.keys(dto.values).length > 0) {
      await this.save({ deviceId: device.id, timestamp, data: dto.values });
    }

    return true;
  }

Add the import at the top of the file:

import type { OrgIngestDto } from '@/telemetry/dto/org-ingest.dto';

Step 4: Run test to verify it passes

Run: make backend-test FILE=backend/src/telemetry/telemetry.service.spec.ts
Expected: PASS (the 3 new cases).

Step 5: Commit

git add backend/src/telemetry/telemetry.service.ts backend/src/telemetry/telemetry.service.spec.ts
git commit -m "feat(telemetry): add per-org ingest with attribute sync"

Task 4: Integration ingest controller + wiring

New route POST /api/v1/integrations/thingsboard/ingest, guarded by OrgIngestGuard, reading the org from the attached connection. The legacy POST /telemetry/tb-ingest controller stays untouched for backward compatibility.

Files:

Create: backend/src/telemetry/controllers/integration-ingest.controller.ts
Modify: backend/src/telemetry/telemetry.module.ts
Step 1: Write the controller

// backend/src/telemetry/controllers/integration-ingest.controller.ts
import { Body, Controller, HttpCode, HttpStatus, Post, Req, UseGuards } from '@nestjs/common';
import { ApiOperation, ApiResponse, ApiTags } from '@nestjs/swagger';
import type { IntegrationConnection } from '@/integrations/entities/integration-connection.entity';
import { OrgIngestDto } from '@/telemetry/dto/org-ingest.dto';
import { OrgIngestGuard } from '@/telemetry/guards/org-ingest.guard';
import { TelemetryService } from '@/telemetry/telemetry.service';

@ApiTags('Telemetry')
@Controller('integrations/thingsboard')
export class IntegrationIngestController {
  constructor(private readonly telemetryService: TelemetryService) {}

  @Post('ingest')
  @UseGuards(OrgIngestGuard)
  @HttpCode(HttpStatus.ACCEPTED)
  @ApiOperation({ summary: 'Per-org ThingsBoard rule-chain ingest' })
  @ApiResponse({ status: 202, description: 'Accepted' })
  @ApiResponse({ status: 401, description: 'Invalid or missing ingest token' })
  async ingest(
    @Req() request: { integrationConnection: IntegrationConnection },
    @Body() dto: OrgIngestDto,
  ): Promise<{ accepted: boolean }> {
    const accepted = await this.telemetryService.ingestForOrg(
      request.integrationConnection.organizationId,
      dto,
    );
    return { accepted };
  }
}

Step 2: Wire into telemetry.module.ts

Import IntegrationsModule (for the IntegrationConnection repo used by the guard and the DeviceIntegrationService used later), register the controller and guard. In backend/src/telemetry/telemetry.module.ts:

Add to imports: IntegrationsModule (which exports TypeOrmModule for IntegrationConnection). Add TypeOrmModule.forFeature([IntegrationConnection]) if the repo is not exported for injection in the guard - prefer importing IntegrationsModule and having it export TypeOrmModule.
Add IntegrationIngestController to controllers.
Add OrgIngestGuard to providers.

import { IntegrationsModule } from '@/integrations/integrations.module';
import { IntegrationIngestController } from '@/telemetry/controllers/integration-ingest.controller';
import { OrgIngestGuard } from '@/telemetry/guards/org-ingest.guard';

Step 3: Verify build + boot

Run: cd backend && bun run build
Expected: compiles.

Step 4: Manual smoke (optional but recommended)

With the app running and a configured+connected org (from Plan 2), POST a telemetry sample:

curl -s -X POST http://localhost:3000/api/v1/integrations/thingsboard/ingest \
  -H 'Content-Type: application/json' \
  -H 'x-ingest-token: <ingestToken from the connection>' \
  -d '{"deviceName":"<a serialNumber in that org>","values":{"ph":7.1}}'

Expected: 202 {"accepted":true}; a new device_telemetry row and telemetry:latest:{deviceId} Redis key.

Step 5: Commit

git add backend/src/telemetry/controllers/integration-ingest.controller.ts backend/src/telemetry/telemetry.module.ts
git commit -m "feat(telemetry): add per-org integration ingest webhook"

Task 5: Decouple `getLatest` - read from internal DB

Replace the TB fallback with a device_telemetry latest-row read. The Redis-hit path is unchanged.

Files:

Modify: backend/src/telemetry/telemetry.service.ts (getLatest, lines 118-162)
Test: backend/src/telemetry/telemetry.service.spec.ts
Step 1: Write the failing test

// in telemetry.service.spec.ts
describe('getLatest (internal DB)', () => {
  it('returns the cached value on a Redis hit', async () => {
    redisService.get.mockResolvedValue(
      JSON.stringify({ time: '2026-07-11T00:00:00Z', deviceId: 'dev-1', data: { ph: 7 } }),
    );
    const result = await service.getLatest('dev-1');
    expect(result?.data).toEqual({ ph: 7 });
    expect(telemetryRepository.createQueryBuilder).not.toHaveBeenCalled();
  });

  it('reads the latest device_telemetry row on a cache miss (no TB call)', async () => {
    redisService.get.mockResolvedValue(null);
    telemetryRepository.findOne = jest.fn().mockResolvedValue({
      time: new Date('2026-07-11T00:00:00Z'),
      deviceId: 'dev-1',
      data: { ph: 6.9 },
      signalStrength: -50,
      batteryLevel: 80,
    });
    const result = await service.getLatest('dev-1');
    expect(result).toEqual(
      expect.objectContaining({ deviceId: 'dev-1', data: { ph: 6.9 } }),
    );
  });

  it('returns null when there is no telemetry', async () => {
    redisService.get.mockResolvedValue(null);
    telemetryRepository.findOne = jest.fn().mockResolvedValue(null);
    expect(await service.getLatest('dev-1')).toBeNull();
  });
});

Step 2: Run test to verify it fails

Run: make backend-test FILE=backend/src/telemetry/telemetry.service.spec.ts
Expected: FAIL - current code calls TB, not telemetryRepository.findOne.

Step 3: Rewrite getLatest

Replace the TB fallback block (from const thingsboardDeviceId = await this.resolveTbDeviceId(deviceId); to the end of the method body) with a DB read:

  async getLatest(deviceId: string): Promise<TelemetryLatestDto | null> {
    try {
      const cacheKey = `telemetry:latest:${deviceId}`;
      const cached = await this.redisService.get(cacheKey);
      if (cached) {
        return JSON.parse(cached);
      }

      const row = await this.telemetryRepository.findOne({
        where: { deviceId },
        order: { time: 'DESC' },
      });
      if (!row) {
        return null;
      }

      const data = row.data as Record<string, number | string | boolean>;
      const response: TelemetryLatestDto = {
        time: row.time.toISOString(),
        deviceId,
        data,
        signalStrength: row.signalStrength ?? undefined,
        batteryLevel: row.batteryLevel ?? undefined,
      };

      await this.redisService.setex(cacheKey, 300, JSON.stringify(response));
      return response;
    } catch (error) {
      this.logger.error(`Failed to get latest telemetry for device ${deviceId}:`, error);
      throw error;
    }
  }

Step 4: Run test to verify it passes

Run: make backend-test FILE=backend/src/telemetry/telemetry.service.spec.ts
Expected: PASS.

Step 5: Commit

git add backend/src/telemetry/telemetry.service.ts backend/src/telemetry/telemetry.service.spec.ts
git commit -m "refactor(telemetry): read latest from internal DB instead of ThingsBoard"

Task 6: Decouple `getHistory`, `getPondTelemetry`, `exportTelemetry`

All three query device_telemetry by internal deviceId over a time range and reuse the existing pure transforms (mergeTbSeries becomes a jsonb-row mapper; keep calculateAggregation). Model on devices.service.ts:601-669.

Files:

Modify: backend/src/telemetry/telemetry.service.ts (getHistory 164-198, getPondTelemetry 200-242, exportTelemetry 451-484, remove resolveTbDeviceId/resolveKeys/TB-specific helpers no longer used)
Test: backend/src/telemetry/telemetry.service.spec.ts
Step 1: Write the failing tests

// in telemetry.service.spec.ts
describe('getHistory (internal DB)', () => {
  function mockRows(rows: Array<{ time: Date; data: Record<string, unknown> }>) {
    const qb = {
      where: jest.fn().mockReturnThis(),
      andWhere: jest.fn().mockReturnThis(),
      orderBy: jest.fn().mockReturnThis(),
      getMany: jest.fn().mockResolvedValue(rows.map((r) => ({ deviceId: 'dev-1', ...r }))),
    };
    telemetryRepository.createQueryBuilder = jest.fn().mockReturnValue(qb);
    return qb;
  }

  it('returns data points from device_telemetry (no TB call)', async () => {
    mockRows([
      { time: new Date('2026-07-11T00:00:00Z'), data: { ph: 7, do: 6 } },
      { time: new Date('2026-07-11T01:00:00Z'), data: { ph: 7.2, do: 6.1 } },
    ]);
    const result = await service.getHistory('dev-1', { parameters: ['ph'] });
    expect(result.data).toHaveLength(2);
    expect(result.data[0]).toEqual(expect.objectContaining({ ph: 7 }));
    expect(result.aggregated.ph).toBeDefined();
  });

  it('returns empty data when there are no rows', async () => {
    mockRows([]);
    const result = await service.getHistory('dev-1', {});
    expect(result).toEqual({ data: [], aggregated: {} });
  });
});

(Add analogous cases: getPondTelemetry queries by the pond’s device ids; exportTelemetry produces a CSV StreamableFile from DB rows.)

Step 2: Run test to verify it fails

Run: make backend-test FILE=backend/src/telemetry/telemetry.service.spec.ts
Expected: FAIL.

Step 3: Rewrite the three methods

Introduce one private helper reused by all three, and rewrite each method to call it. Add:

  private async readDeviceHistory(
    deviceId: string,
    from: Date,
    to: Date,
    parameters?: string[],
  ): Promise<TelemetryDataPoint[]> {
    const rows = await this.telemetryRepository
      .createQueryBuilder('telemetry')
      .where('telemetry.deviceId = :deviceId', { deviceId })
      .andWhere('telemetry.time >= :from', { from })
      .andWhere('telemetry.time <= :to', { to })
      .orderBy('telemetry.time', 'ASC')
      .getMany();

    return rows.map((row) => {
      const point: TelemetryDataPoint = { time: row.time.toISOString() };
      const data = row.data as Record<string, number | string | boolean>;
      const keys = parameters && parameters.length > 0 ? parameters : Object.keys(data);
      for (const key of keys) {
        if (key in data) {
          point[key] = data[key];
        }
      }
      return point;
    });
  }

getHistory:

  async getHistory(
    deviceId: string,
    query: TelemetryHistoryQueryDto,
  ): Promise<TelemetryHistoryResponseDto> {
    try {
      const from = query.from ? new Date(query.from) : new Date(Date.now() - 24 * 60 * 60 * 1000);
      const to = query.to ? new Date(query.to) : new Date();
      const data = await this.readDeviceHistory(deviceId, from, to, query.parameters);
      if (data.length === 0) {
        return { data: [], aggregated: {} };
      }
      const aggregated = this.calculateAggregation(data, query.parameters);
      return { data, aggregated };
    } catch (error) {
      this.logger.error(`Failed to get telemetry history for device ${deviceId}:`, error);
      throw error;
    }
  }

getPondTelemetry (query all devices in the pond, drop the thingsboardDeviceId filter):

  async getPondTelemetry(
    pondId: string,
    query: TelemetryHistoryQueryDto,
  ): Promise<TelemetryHistoryResponseDto> {
    try {
      const devices = await this.deviceRepository.find({ where: { pondId } });
      if (devices.length === 0) {
        return { data: [], aggregated: {} };
      }
      const from = query.from ? new Date(query.from) : new Date(Date.now() - 24 * 60 * 60 * 1000);
      const to = query.to ? new Date(query.to) : new Date();

      const data: TelemetryDataPoint[] = [];
      for (const device of devices) {
        const points = await this.readDeviceHistory(device.id, from, to, query.parameters);
        data.push(...points);
      }
      data.sort((a, b) => (a.time < b.time ? -1 : a.time > b.time ? 1 : 0));
      const aggregated = this.calculateAggregation(data, query.parameters);
      return { data, aggregated };
    } catch (error) {
      this.logger.error(`Failed to get pond telemetry for pond ${pondId}:`, error);
      throw error;
    }
  }

exportTelemetry:

  async exportTelemetry(query: ExportTelemetryQueryDto): Promise<StreamableFile> {
    const from = new Date(query.from);
    const to = new Date(query.to);

    const points: TelemetryDataPoint[] = [];
    for (const deviceId of query.deviceIds) {
      points.push(...(await this.readDeviceHistory(deviceId, from, to, query.parameters)));
    }
    points.sort((a, b) => (a.time < b.time ? -1 : a.time > b.time ? 1 : 0));
    const csvContent = this.generateCSV(points, query.parameters);

    const buffer = Buffer.from(csvContent, 'utf-8');
    const filename = `telemetry_export_${Date.now()}.csv`;
    return new StreamableFile(buffer, {
      type: 'text/csv',
      disposition: `attachment; filename="${filename}"`,
    });
  }

Delete the now-unused resolveTbDeviceId, resolveKeys, mergeTbSeries, buildAggregationOptions, parseIntervalToMs, and coerceTelemetryValue if and only if nothing else references them (grep first). Keep calculateAggregation and generateCSV.

Step 4: Run test to verify it passes

Run: make backend-test FILE=backend/src/telemetry/telemetry.service.spec.ts
Expected: PASS.

Step 5: Commit

git add backend/src/telemetry/telemetry.service.ts backend/src/telemetry/telemetry.service.spec.ts
git commit -m "refactor(telemetry): serve history/pond/export from internal DB"

Task 7: Drop `ThingsBoardClientService` from `TelemetryService`

With reads decoupled, TelemetryService no longer needs TB.

Files:

Modify: backend/src/telemetry/telemetry.service.ts (remove the thingsBoardClientService constructor param and import)
Modify: backend/src/telemetry/telemetry.module.ts (remove ThingsboardModule import if present)
Step 1: Remove the injection

Delete private thingsBoardClientService: ThingsBoardClientService, from the constructor and its import. Also remove the legacy ingestFromThingsBoard method and the legacy TbIngestDto import IF the legacy tb-ingest route is being retired; per the design we KEEP the legacy route for backward compatibility, so leave ingestFromThingsBoard and the legacy controller in place - it uses save() and the deviceRepository only, no TB client. Confirm ingestFromThingsBoard does not reference thingsBoardClientService (it does not) before removing the injection.

Step 2: Verify build + tests

Run: cd backend && bun run build && make backend-test FILE=backend/src/telemetry
Expected: compiles; telemetry specs PASS.

Step 3: Commit

git add backend/src/telemetry/telemetry.service.ts backend/src/telemetry/telemetry.module.ts
git commit -m "refactor(telemetry): drop ThingsBoard client dependency"

Task 8: Internal telemetry bus + monitoring live-stream migration

Feed the /monitoring stream from an in-process event bus emitted by save() instead of the TB WebSocket.

Files:

Create: backend/src/telemetry/internal-telemetry-bus.ts
Modify: backend/src/telemetry/telemetry.service.ts (emit in save), telemetry.module.ts (provide+export the bus)
Modify: backend/src/monitoring/services/monitoring-stream.service.ts (subscribe to the bus), monitoring.module.ts
Test: backend/src/telemetry/internal-telemetry-bus.spec.ts
Step 1: Write the failing bus test

// backend/src/telemetry/internal-telemetry-bus.spec.ts
import { InternalTelemetryBus } from '@/telemetry/internal-telemetry-bus';

describe('InternalTelemetryBus', () => {
  it('delivers published events to subscribers and stops after unsubscribe', () => {
    const bus = new InternalTelemetryBus();
    const received: unknown[] = [];
    const unsubscribe = bus.subscribe((e) => received.push(e));

    bus.publish({ deviceId: 'dev-1', data: { ph: 7 }, timestamp: new Date(0) });
    expect(received).toHaveLength(1);

    unsubscribe();
    bus.publish({ deviceId: 'dev-1', data: { ph: 8 }, timestamp: new Date(0) });
    expect(received).toHaveLength(1);
  });
});

Step 2: Run test to verify it fails

Run: make backend-test FILE=backend/src/telemetry/internal-telemetry-bus.spec.ts
Expected: FAIL - cannot find module.

Step 3: Write the bus

// backend/src/telemetry/internal-telemetry-bus.ts
import { Injectable } from '@nestjs/common';

export interface InternalTelemetryEvent {
  deviceId: string;
  data: Record<string, unknown>;
  timestamp: Date;
}

export type TelemetryListener = (event: InternalTelemetryEvent) => void;

/**
 * In-process pub/sub for telemetry ingested through TelemetryService.save().
 * The monitoring live-stream subscribes here instead of the ThingsBoard WS.
 */
@Injectable()
export class InternalTelemetryBus {
  private readonly listeners = new Set<TelemetryListener>();

  subscribe(listener: TelemetryListener): () => void {
    this.listeners.add(listener);
    return () => this.listeners.delete(listener);
  }

  publish(event: InternalTelemetryEvent): void {
    for (const listener of this.listeners) {
      listener(event);
    }
  }
}

Step 4: Emit from save()

Inject InternalTelemetryBus into TelemetryService and, in save() right after the existing this.websocketGateway.emitTelemetry(dto.deviceId, dto.data); line (telemetry.service.ts:75), add:

    this.internalTelemetryBus.publish({
      deviceId: dto.deviceId,
      data: dto.data,
      timestamp,
    });

Provide + export InternalTelemetryBus from telemetry.module.ts.

Step 5: Repoint the monitoring stream

In monitoring-stream.service.ts, replace the TB adapter subscription. The report identifies openDeviceSubscriptions (110-164) calling this.adapter.subscribeToScope(...) and attachClient calling this.adapter.registerConnectionHooks(...). Change the constructor to inject InternalTelemetryBus instead of ThingsboardIntegrationAdapterService, and rewrite the per-device subscription to filter bus events by the scope’s internal device ids, mapping each InternalTelemetryEvent to MonitoringEventDto (source: 'internal'). Keep attachClient’s listener set, evictDevice, and the MonitoringEventDto output contract identical - the gateway is unaffected.

Write a focused unit test monitoring-stream.service.spec.ts asserting that publishing a bus event for a subscribed device delivers a MonitoringEventDto to the attached listener, and that events for non-subscribed devices are filtered out. (Model the event→DTO mapping on the existing normalize, but from InternalTelemetryEvent: emit one MonitoringEventDto per metric key in data.)

Step 6: Run tests

Run: make backend-test FILE=backend/src/telemetry/internal-telemetry-bus.spec.ts
Run: make backend-test FILE=backend/src/monitoring/services/monitoring-stream.service.spec.ts
Expected: PASS.

Step 7: Commit

git add backend/src/telemetry/internal-telemetry-bus.ts backend/src/telemetry/internal-telemetry-bus.spec.ts backend/src/telemetry/telemetry.service.ts backend/src/telemetry/telemetry.module.ts backend/src/monitoring/services/monitoring-stream.service.ts backend/src/monitoring/services/monitoring-stream.service.spec.ts backend/src/monitoring/monitoring.module.ts
git commit -m "feat(monitoring): drive live stream from internal telemetry bus"

Task 9: Monitoring snapshot/history + health from internal data

Files:

Modify: backend/src/monitoring/services/monitoring-query.service.ts (getSnapshot 27-92, getHistory 94-137)
Modify: backend/src/monitoring/services/monitoring-health.service.ts (connected, line 26)
Test: corresponding .spec.ts
Step 1: Rewrite reads to query device_telemetry

Inject the DeviceTelemetry repository into MonitoringQueryService. getSnapshot: per device in scope, read the latest device_telemetry row (like Task 5) and emit MonitoringEventDto per metric with source: 'internal'. getHistory: read the time range for the device (like the Task 6 helper) and build the history points. Remove the ThingsboardIntegrationAdapterService dependency from this service.

Step 2: Replace the health connected signal

In monitoring-health.service.ts, replace connected: this.adapter.isConnected() with an internal signal. Simplest correct choice: connected: true when the internal stream/observability pipeline is healthy (the module is always internally connected once booted). Remove the adapter dependency. Update the spec to assert the new source.

Step 3: Write/adjust unit tests

Add cases to monitoring-query.service.spec.ts proving snapshot/history come from the telemetry repo (no adapter calls). Adjust monitoring-health.service.spec.ts.

Step 4: Run tests

Run: make backend-test FILE=backend/src/monitoring
Expected: PASS.

Step 5: Commit

git add backend/src/monitoring/services/monitoring-query.service.ts backend/src/monitoring/services/monitoring-query.service.spec.ts backend/src/monitoring/services/monitoring-health.service.ts backend/src/monitoring/services/monitoring-health.service.spec.ts backend/src/monitoring/monitoring.module.ts
git commit -m "refactor(monitoring): serve snapshot/history/health from internal data"

Task 10: RPC via connector (device command dispatch + monitoring command)

Command egress still travels through TB, but via the connector, resolving the external id from device_integrations instead of device.thingsboardDeviceId.

Files:

Modify: backend/src/devices/device-command-dispatch.service.ts
Modify: backend/src/monitoring/services/monitoring-command.service.ts
Modify: backend/src/devices/devices.module.ts, backend/src/monitoring/monitoring.module.ts (import IntegrationsModule)
Test: the respective .spec.ts
Step 1: Rewrite device-command-dispatch.service.ts

Inject ConnectorConnectionManager and DeviceIntegrationService (both exported by IntegrationsModule) instead of ThingsBoardClientService. Resolve the external id and dispatch:

async dispatchEndpointValue(input: {
  targetDeviceEndpointId: string;
  targetValue: unknown;
}): Promise<unknown> {
  const endpoint = await this.loadEndpoint(input.targetDeviceEndpointId); // existing load logic
  const device = endpoint.device;
  const externalId = await this.deviceIntegrationService.resolveExternalId(device.id, 'thingsboard');
  if (!externalId) {
    throw new BadRequestException('Device is not mapped to a ThingsBoard integration');
  }
  const binding = endpoint.profileEndpoint.transportBinding;
  const connector = await this.connectionManager.getConnector(
    device.organizationId as string,
    'thingsboard',
  );
  return connector.sendRpc(externalId, {
    method: binding.method,
    params: { [binding.paramKey]: input.targetValue },
  });
}

Update the unit test to mock deviceIntegrationService.resolveExternalId and connectionManager.getConnector (returning a connector whose sendRpc is a jest mock) and assert sendRpc is called with the resolved external id.

Step 2: Rewrite monitoring-command.service.ts dispatch

In dispatch (252-326), replace the device.thingsboardDeviceId resolution + this.adapter.sendDeviceRpc(...) with deviceIntegrationService.resolveExternalId(device.id) + connectionManager.getConnector(organizationId).sendRpc(externalId, { method, params }). Preserve the correlation-id/record behavior. Update its spec accordingly.

Step 3: Run tests

Run: make backend-test FILE=backend/src/devices/device-command-dispatch.service.spec.ts
Run: make backend-test FILE=backend/src/monitoring/services/monitoring-command.service.spec.ts
Expected: PASS.

Step 4: Commit

git add backend/src/devices/device-command-dispatch.service.ts backend/src/monitoring/services/monitoring-command.service.ts backend/src/devices/devices.module.ts backend/src/monitoring/monitoring.module.ts backend/src/devices/**/*.spec.ts backend/src/monitoring/**/*.spec.ts
git commit -m "refactor(commands): dispatch RPC through the connector layer"

Task 11: Device provisioning/removal via connector + mapping

DevicesService.create pushes new devices to TB via the connector and stores the mapping in device_integrations; remove and the system-inventory hard delete use the connector to delete the remote device.

Files:

Modify: backend/src/devices/devices.service.ts (create provisioning 236-325, provisionThingsBoardDevice 381-390, rollbackThingsBoardDevice 412-419, remove 501-534)
Modify: backend/src/devices/system-device-inventory.service.ts (deleteThingsBoardDeviceBestEffort 266-283)
Modify: backend/src/devices/devices.module.ts
Test: backend/src/devices/*.spec.ts
Step 1: Rewrite provisioning to use the connector + mapping

Inject ConnectorConnectionManager and DeviceIntegrationService. provisionThingsBoardDevice(serialNumber, organizationId) becomes:

private async provisionRemoteDevice(
  serialNumber: string,
  organizationId: string,
): Promise<{ externalId: string; accessToken: string } | null> {
  const connection = await this.integrationConnectionService.getForOrg(organizationId, 'thingsboard');
  if (!connection || connection.status !== IntegrationStatus.CONNECTED) {
    return null; // org has no active TB integration; device is HMP-only
  }
  const connector = await this.connectionManager.getConnector(organizationId, 'thingsboard');
  const { externalId } = await connector.pushDevice({ name: serialNumber });
  const { accessToken } = await connector.getRemoteDeviceCredentials(externalId);
  return { externalId, accessToken };
}

After the device row is persisted, write the mapping via deviceIntegrationService.upsertMapping({ deviceId, organizationId, provider: 'thingsboard', externalId, accessToken }) instead of setting device.thingsboardDeviceId/accessToken. Rollback becomes connector.deleteRemoteDevice(externalId). Preserve the name-conflict retry semantics (the connector still throws ThingsBoardApiError with tbErrorCode).

Important behavior change to note in the plan: provisioning is now conditional on the org having a connected TB integration. Orgs with no integration get HMP-only devices (no remote push) - this is the correct plugin behavior. Confirm this matches product expectations during review.

Step 2: Rewrite remove and system-inventory delete

In remove, replace rollbackThingsBoardDevice(device.thingsboardDeviceId) with: resolve externalId via deviceIntegrationService.resolveExternalId(device.id), and if present, connector.deleteRemoteDevice(externalId) (best-effort) then deviceIntegrationService.deleteMapping(device.id). Keep the monitoringStreamService.evictDevice(id) call. Apply the same pattern in system-device-inventory.service.ts.

Step 3: Update devices.module.ts

Import IntegrationsModule; remove ThingsboardModule. Ensure IntegrationConnectionService (from Plan 2) is reachable - export it from IntegrationsModule (add to that module’s exports).

Step 4: Update unit tests

Adjust devices.service.spec.ts (and system-device-inventory.service.spec.ts) to mock the connector/manager/mapping/connection service. Add a case: creating a device for an org WITHOUT a connected integration persists the device and creates NO mapping and calls no connector.

Step 5: Run tests

Run: make backend-test FILE=backend/src/devices
Expected: PASS.

Step 6: Commit

git add backend/src/devices/devices.service.ts backend/src/devices/system-device-inventory.service.ts backend/src/devices/devices.module.ts backend/src/devices/**/*.spec.ts
git commit -m "refactor(devices): provision and delete remote devices via the connector"

Task 12: Offline detection from `lastSeenAt`

Replace the TB polling cron with internal offline detection.

Files:

Modify: backend/src/devices/device-activity-monitor.service.ts
Test: backend/src/devices/device-activity-monitor.service.spec.ts
Step 1: Rewrite the cron

Remove the ThingsBoardClientService dependency and the getDeviceActivity call. The cron now marks a device offline when lastSeenAt is older than a threshold (default 5 minutes, aligning with the old cadence), and keeps online otherwise. Raise the offline alert exactly as before when a device transitions to offline.

private readonly offlineThresholdMs = 5 * 60 * 1000;

@Cron(CronExpression.EVERY_5_MINUTES)
async checkDeviceActivity(): Promise<void> {
  const devices = await this.deviceRepository.find({
    where: { isActive: true, deletedAt: IsNull() },
  });
  const now = Date.now();
  for (const device of devices) {
    const lastSeen = device.lastSeenAt ? device.lastSeenAt.getTime() : 0;
    const isOnline = now - lastSeen <= this.offlineThresholdMs;
    const nextStatus = isOnline ? 'online' : 'offline';
    if (device.status !== nextStatus) {
      await this.deviceRepository.update(device.id, { status: nextStatus });
      if (nextStatus === 'offline') {
        await this.raiseOfflineAlert(device);
      }
    }
  }
}

Step 2: Update the unit test

Assert: a device with a stale lastSeenAt transitions to offline and raises an alert; a fresh one stays online; no TB client is called.

Step 3: Run tests

Run: make backend-test FILE=backend/src/devices/device-activity-monitor.service.spec.ts
Expected: PASS.

Step 4: Commit

git add backend/src/devices/device-activity-monitor.service.ts backend/src/devices/device-activity-monitor.service.spec.ts
git commit -m "refactor(devices): derive offline status from lastSeenAt, drop TB polling"

Task 13: Data migration + drop legacy columns

Backfill device_integrations from the existing devices.thingsboard_device_id/access_token, then drop the legacy columns.

Files:

Create: backend/src/migrations/1778940000000-BackfillDeviceIntegrationsFromDevices.ts
Create: backend/src/migrations/1778950000000-DropLegacyDeviceTbColumns.ts
Modify: backend/src/devices/entities/device.entity.ts
Step 1: Write the backfill migration

Credentials in the old access_token column are plaintext; the new credentials_encrypted column expects ciphertext. Since the migration cannot access the app’s CredentialCipher, backfill external_id only and leave credentials_encrypted NULL; the next resync (Plan 2) re-fetches and encrypts device credentials. Document this in the migration.

// backend/src/migrations/1778940000000-BackfillDeviceIntegrationsFromDevices.ts
import type { MigrationInterface, QueryRunner } from 'typeorm';

export class BackfillDeviceIntegrationsFromDevices1778940000000 implements MigrationInterface {
  name = 'BackfillDeviceIntegrationsFromDevices1778940000000';

  // Backfills external_id from devices.thingsboard_device_id. Access tokens are
  // intentionally NOT copied: the old column is plaintext and device_integrations
  // stores ciphertext. Credentials are repopulated on the next integration resync.
  public async up(queryRunner: QueryRunner): Promise<void> {
    await queryRunner.query(`
      INSERT INTO device_integrations
        (id, device_id, organization_id, provider, external_id, credentials_encrypted, last_synced_at, created_at, updated_at)
      SELECT gen_random_uuid(), d.id, d.organization_id, 'thingsboard',
             d.thingsboard_device_id::text, NULL, now(), now(), now()
      FROM devices d
      WHERE d.thingsboard_device_id IS NOT NULL
        AND d.organization_id IS NOT NULL
        AND NOT EXISTS (
          SELECT 1 FROM device_integrations di
          WHERE di.device_id = d.id AND di.provider = 'thingsboard'
        );
    `);
  }

  public async down(queryRunner: QueryRunner): Promise<void> {
    await queryRunner.query(`DELETE FROM device_integrations WHERE provider = 'thingsboard';`);
  }
}

Step 2: Write the drop-columns migration

// backend/src/migrations/1778950000000-DropLegacyDeviceTbColumns.ts
import type { MigrationInterface, QueryRunner } from 'typeorm';
import { TableColumn } from 'typeorm';

export class DropLegacyDeviceTbColumns1778950000000 implements MigrationInterface {
  name = 'DropLegacyDeviceTbColumns1778950000000';

  public async up(queryRunner: QueryRunner): Promise<void> {
    const hasIndex = await queryRunner.query(
      `SELECT 1 FROM pg_indexes WHERE indexname = 'IDX_devices_thingsboard_device_id'`,
    );
    if (hasIndex.length > 0) {
      await queryRunner.dropIndex('devices', 'IDX_devices_thingsboard_device_id');
    }
    await queryRunner.dropColumn('devices', 'thingsboard_device_id');
    await queryRunner.dropColumn('devices', 'access_token');
    await queryRunner.dropColumn('devices', 'mqtt_topic');
    await queryRunner.dropColumn('devices', 'secret');
  }

  public async down(queryRunner: QueryRunner): Promise<void> {
    await queryRunner.addColumn(
      'devices',
      new TableColumn({ name: 'thingsboard_device_id', type: 'uuid', isNullable: true }),
    );
    await queryRunner.addColumn(
      'devices',
      new TableColumn({ name: 'access_token', type: 'varchar', length: '100', isNullable: true }),
    );
    await queryRunner.addColumn(
      'devices',
      new TableColumn({ name: 'mqtt_topic', type: 'varchar', length: '255', isNullable: true }),
    );
    await queryRunner.addColumn(
      'devices',
      new TableColumn({ name: 'secret', type: 'varchar', length: '100', isNullable: true }),
    );
  }
}

Step 3: Remove the columns from the entity

In backend/src/devices/entities/device.entity.ts, delete the mqttTopic (73-74), thingsboardDeviceId (76-77), accessToken (79-80), and secret (82-83) column declarations, and delete the toJSON() method (142-145) that stripped accessToken/secret (no longer needed). Grep the codebase for any remaining references to these properties and fix them (there should be none after Tasks 5-12; if a spec references them, update it).

Step 4: Run migrations + full build

Run: cd backend && bun run migration:run
Expected: backfill then drop succeed.

Run: cd backend && bun run build
Expected: compiles with no references to the removed properties.

Step 5: Commit

git add backend/src/migrations/1778940000000-BackfillDeviceIntegrationsFromDevices.ts backend/src/migrations/1778950000000-DropLegacyDeviceTbColumns.ts backend/src/devices/entities/device.entity.ts
git commit -m "refactor(devices): backfill device_integrations and drop legacy TB/MQTT columns"

Task 14: Remove the old ThingsBoard module, demote health, clean cruft

Files:

Modify: backend/src/health/health.service.ts
Modify: backend/src/app.module.ts
Delete: backend/src/thingsboard/
Modify: backend/CLAUDE.md, .env.production
Step 1: Demote TB in health

The connector is now per-org and optional, so a global TB login no longer represents platform health. Remove checkThingsBoard from the fatal isHealthy computation (health.service.ts:21). Either drop the check entirely or report it as an informational field that does not affect isHealthy. Update health.service.spec.ts.

Step 2: Confirm no remaining imports of the old client

Run: cd backend && rg "thingsboard-client.service|ThingsBoardClientService|ThingsboardModule|ThingsboardIntegrationAdapterService" src
Expected: no non-deleted references remain except inside src/thingsboard/ itself (about to be deleted) and, if still used, the monitoring adapter. If the monitoring adapter (ThingsboardIntegrationAdapterService) is now unused after Tasks 8-10, delete it too; if any path still needs it, leave it and note why.

Step 3: Remove the module

Delete backend/src/thingsboard/ and remove ThingsboardModule from app.module.ts imports and its import statement.

Step 4: Clean stale docs/config
backend/CLAUDE.md: remove the mqtt/ module row and the “MQTT Integration” section; update the stack line and telemetry/ThingsBoard descriptions to reflect the connector model.
.env.production: remove MQTT_BROKER_URL, MQTT_USERNAME, MQTT_PASSWORD, EMQX_DASHBOARD_USER, EMQX_DASHBOARD_PASSWORD.
Step 5: Full build + test sweep

Run: cd backend && bun run build
Expected: compiles.

Run: make backend-test FILE=backend/src/telemetry and make backend-test FILE=backend/src/devices and make backend-test FILE=backend/src/monitoring and make backend-test FILE=backend/src/integrations and make backend-test FILE=backend/src/health
Expected: all PASS.

Step 6: Commit

git add backend/src/health/health.service.ts backend/src/health/health.service.spec.ts backend/src/app.module.ts backend/CLAUDE.md .env.production
git rm -r backend/src/thingsboard
git commit -m "refactor: remove legacy ThingsBoard module and stale MQTT/EMQX config"

Task 15: Lint + end-to-end verification

Step 1: Biome

Run: bun run check
Expected: clean; fix any issues.

Step 2: Backend e2e

Run: cd backend && bun run test:e2e
Expected: PASS (including the Plan 2 connect e2e and migration-order e2e).

Step 3: Full backend unit run

Run: cd backend && bun run test
Expected: PASS.

Step 4: Manual regression checklist (with a connected org)
Ingest a sample via the per-org webhook (Task 4 curl) → device shows online, telemetry appears in the device history UI (now DB-backed).
Open a device/pond history view → data renders from internal DB.
Trigger a device command → RPC reaches TB via the connector.
Stop the TB container → platform health stays ok; telemetry reads and history still work from internal DB.
Step 5: Commit any lint fixes

git add -A
git commit -m "chore: lint and final verification for plan 3"

Self-Review Notes (author)

Spec coverage: Section 7 (reads from internal DB, monitoring live-stream to internal, connector keeps outbound RPC/push/delete) - Tasks 5, 6, 8, 9, 10, 11. Section 8 (offline from ingest, drop TB poll) - Task 12. Per-org ingest (Section 6 webhook / Section 3 spec) - Tasks 2, 3, 4. Section 4.4 + 11 (cleanup, data migration, column drop, module/env removal, health demote) - Tasks 13, 14.
Type consistency: OrgIngestDto (Task 2) used in Tasks 3, 4. InternalTelemetryEvent/InternalTelemetryBus (Task 8) used in Task 9’s stream. DeviceIntegrationService (Task 1) used in Tasks 10, 11. TelemetryDataPoint/TelemetryHistoryResponseDto/TelemetryLatestDto are existing types reused unchanged (Tasks 5, 6). readDeviceHistory defined once (Task 6) and reused by all three read methods.
Ordering rationale: reads are decoupled (5-7) before dropping columns (13); the connector-based provisioning/RPC (10, 11) and offline detection (12) remove the last thingsboardDeviceId readers before the column drop; the old module is removed last (14) once nothing imports it.
Deliberate behavior change flagged for review: device provisioning is now conditional on a connected org integration (Task 11 Step 1) - orgs without an integration get HMP-only devices. Confirm during review.
Legacy tb-ingest route is intentionally preserved (Task 7 Step 1) for deployed rule chains; only its TB-client-free path remains.
Watch-outs: device_telemetry is not a hypertable - large history ranges are plain scans (acceptable now, follow-up to enable create_hypertable + continuous aggregates); the backfill migration copies external ids only (credentials re-fetched on resync); confirm the monitoring adapter can be fully deleted (Task 14 Step 2) or document why it stays.
Migration timestamps 1778940000000, 1778950000000 follow Plan 2’s 1778930000000.

ThingsBoard Connector Plugin - Ingest & Telemetry Decoupling (Plan 3 of 3) Implementation Plan

File Structure

Task 1: DeviceIntegrationService - resolve external id / credentials

Task 2: Per-org ingest DTO + guard

Task 3: TelemetryService.ingestForOrg + attribute sync

Task 4: Integration ingest controller + wiring

Task 5: Decouple getLatest - read from internal DB

Task 6: Decouple getHistory, getPondTelemetry, exportTelemetry

Task 7: Drop ThingsBoardClientService from TelemetryService

Task 8: Internal telemetry bus + monitoring live-stream migration

Task 9: Monitoring snapshot/history + health from internal data

Task 10: RPC via connector (device command dispatch + monitoring command)

Task 11: Device provisioning/removal via connector + mapping

Task 12: Offline detection from lastSeenAt

Task 13: Data migration + drop legacy columns

Task 14: Remove the old ThingsBoard module, demote health, clean cruft

Task 15: Lint + end-to-end verification

Self-Review Notes (author)

onThisPage

Task 1: `DeviceIntegrationService` - resolve external id / credentials

Task 3: `TelemetryService.ingestForOrg` + attribute sync

Task 5: Decouple `getLatest` - read from internal DB

Task 6: Decouple `getHistory`, `getPondTelemetry`, `exportTelemetry`

Task 7: Drop `ThingsBoardClientService` from `TelemetryService`

Task 12: Offline detection from `lastSeenAt`