ThingsBoard Connector Plugin - Foundation (Plan 1 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: Build the provider-neutral connector foundation - encryption, per-org connection + device-mapping entities, the DeviceConnector interface, a registry, a ThingsBoard implementation, and a per-org connection manager - without touching existing consumers.

Architecture: A new backend/src/integrations/ module. A DeviceConnector interface abstracts provider operations. A ConnectorRegistry maps a provider string to a factory. A ConnectorConnectionManager loads an organization’s encrypted config from the new integration_connection table, decrypts credentials, and returns a cached, config-bound connector. ThingsBoard is the first implementation; its REST logic is ported from the existing thingsboard-client.service.ts. This plan is purely additive: the old ThingsboardModule and all current consumers keep working unchanged.

Tech Stack: NestJS 11, TypeORM 0.3 (Postgres), Node crypto (AES-256-GCM), Jest.

Design spec: docs/superpowers/specs/2026-07-11-thingsboard-connector-plugin-design.md (Sections 3, 4, plus 3.2).

Plan sequence: Plan 1 (this) = foundation. Plan 2 = connection lifecycle/state machine, discovery & mapping, RBAC, frontend. Plan 3 = per-org ingest webhook, telemetry read decoupling, monitoring live-stream migration, legacy cleanup.

File Structure

Create:

backend/src/integrations/crypto/credential-cipher.ts - AES-256-GCM encrypt/decrypt of credential strings.
backend/src/integrations/crypto/credential-cipher.spec.ts
backend/src/integrations/entities/integration-connection.entity.ts - per-org connection config + status.
backend/src/integrations/entities/device-integration.entity.ts - provider-neutral device mapping.
backend/src/integrations/enums/integration-status.enum.ts - connection state machine values.
backend/src/integrations/connectors/device-connector.interface.ts - interface + shared types.
backend/src/integrations/connectors/connector-registry.ts - provider → factory map.
backend/src/integrations/connectors/connector-registry.spec.ts
backend/src/integrations/connectors/thingsboard/thingsboard.connector.ts - TB DeviceConnector impl (ported REST logic).
backend/src/integrations/connectors/thingsboard/thingsboard.connector.spec.ts
backend/src/integrations/connector-connection.manager.ts - per-org connector resolution + cache.
backend/src/integrations/connector-connection.manager.spec.ts
backend/src/integrations/integrations.module.ts - module wiring + TB factory registration.
backend/src/migrations/1778910000000-CreateIntegrationConnection.ts
backend/src/migrations/1778920000000-CreateDeviceIntegration.ts

Modify:

backend/src/app.module.ts:90-132 - register the two new entities.
backend/.env.example - add INTEGRATION_ENCRYPTION_KEY.

Not touched in this plan: devices/, telemetry/, monitoring/, thingsboard/ (existing), Device entity. Those move in Plans 2 and 3.

Task 1: Credential encryption utility

Provider credentials (org login, device access tokens) are stored encrypted. AES-256-GCM with a 32-byte key from INTEGRATION_ENCRYPTION_KEY (base64). Output format: base64(iv).base64(authTag).base64(ciphertext).

Files:

Create: backend/src/integrations/crypto/credential-cipher.ts
Test: backend/src/integrations/crypto/credential-cipher.spec.ts
Step 1: Write the failing test

// backend/src/integrations/crypto/credential-cipher.spec.ts
import { CredentialCipher } from '@/integrations/crypto/credential-cipher';

describe('CredentialCipher', () => {
  // 32 bytes, base64-encoded
  const key = Buffer.alloc(32, 7).toString('base64');
  let cipher: CredentialCipher;

  beforeEach(() => {
    cipher = new CredentialCipher(key);
  });

  it('round-trips a plaintext value', () => {
    const plain = 'super-secret-password';
    const encrypted = cipher.encrypt(plain);
    expect(encrypted).not.toContain(plain);
    expect(cipher.decrypt(encrypted)).toBe(plain);
  });

  it('produces a different ciphertext each call (random IV)', () => {
    expect(cipher.encrypt('same')).not.toBe(cipher.encrypt('same'));
  });

  it('throws when the key is not 32 bytes', () => {
    expect(() => new CredentialCipher(Buffer.alloc(16, 1).toString('base64'))).toThrow(
      /32 bytes/,
    );
  });

  it('throws when decrypting a tampered payload', () => {
    const encrypted = cipher.encrypt('value');
    const tampered = `${encrypted.slice(0, -2)}AA`;
    expect(() => cipher.decrypt(tampered)).toThrow();
  });
});

Step 2: Run test to verify it fails

Run: make backend-test FILE=backend/src/integrations/crypto/credential-cipher.spec.ts
Expected: FAIL - cannot find module credential-cipher.

Step 3: Write minimal implementation

// backend/src/integrations/crypto/credential-cipher.ts
import { createCipheriv, createDecipheriv, randomBytes } from 'node:crypto';

const ALGORITHM = 'aes-256-gcm';
const IV_LENGTH = 12;

/**
 * Encrypts/decrypts short credential strings with AES-256-GCM.
 * Payload format: base64(iv).base64(authTag).base64(ciphertext)
 */
export class CredentialCipher {
  private readonly key: Buffer;

  constructor(base64Key: string) {
    const key = Buffer.from(base64Key, 'base64');
    if (key.length !== 32) {
      throw new Error('INTEGRATION_ENCRYPTION_KEY must decode to 32 bytes (AES-256).');
    }
    this.key = key;
  }

  encrypt(plaintext: string): string {
    const iv = randomBytes(IV_LENGTH);
    const cipher = createCipheriv(ALGORITHM, this.key, iv);
    const ciphertext = Buffer.concat([cipher.update(plaintext, 'utf8'), cipher.final()]);
    const authTag = cipher.getAuthTag();
    return [
      iv.toString('base64'),
      authTag.toString('base64'),
      ciphertext.toString('base64'),
    ].join('.');
  }

  decrypt(payload: string): string {
    const [ivB64, tagB64, dataB64] = payload.split('.');
    if (!ivB64 || !tagB64 || !dataB64) {
      throw new Error('Malformed encrypted payload.');
    }
    const decipher = createDecipheriv(ALGORITHM, this.key, Buffer.from(ivB64, 'base64'));
    decipher.setAuthTag(Buffer.from(tagB64, 'base64'));
    return Buffer.concat([
      decipher.update(Buffer.from(dataB64, 'base64')),
      decipher.final(),
    ]).toString('utf8');
  }
}

Step 4: Run test to verify it passes

Run: make backend-test FILE=backend/src/integrations/crypto/credential-cipher.spec.ts
Expected: PASS (4 tests).

Step 5: Add the env var

Add to backend/.env.example (near the THINGSBOARD_* block):

# Base64-encoded 32-byte key for encrypting integration credentials at rest.
# Generate with: node -e "console.log(require('crypto').randomBytes(32).toString('base64'))"
INTEGRATION_ENCRYPTION_KEY=

Step 6: Commit

git add backend/src/integrations/crypto backend/.env.example
git commit -m "feat(integrations): add AES-256-GCM credential cipher"

Task 2: Integration status enum + `integration_connection` entity

Files:

Create: backend/src/integrations/enums/integration-status.enum.ts
Create: backend/src/integrations/entities/integration-connection.entity.ts
Step 1: Write the status enum

// backend/src/integrations/enums/integration-status.enum.ts
export enum IntegrationStatus {
  DISCONNECTED = 'disconnected',
  CONNECTING = 'connecting',
  SYNCING = 'syncing',
  CONNECTED = 'connected',
  ERROR = 'error',
}

Step 2: Write the entity

Follow the project convention: explicit type on nullable columns, snake_case DB names, organization_id scoping.

// backend/src/integrations/entities/integration-connection.entity.ts
import {
  Column,
  CreateDateColumn,
  Entity,
  Index,
  JoinColumn,
  ManyToOne,
  PrimaryGeneratedColumn,
  UpdateDateColumn,
} from 'typeorm';
import { IntegrationStatus } from '@/integrations/enums/integration-status.enum';
import { Organization } from '@/organizations/entities/organization.entity';

@Entity('integration_connections')
@Index('UQ_integration_connections_org_provider', ['organizationId', 'provider'], {
  unique: true,
})
export class IntegrationConnection {
  @PrimaryGeneratedColumn('uuid')
  id: string;

  @Column({ name: 'organization_id', type: 'uuid' })
  organizationId: string;

  @Column({ type: 'varchar', length: 50 })
  provider: string;

  @Column({ name: 'base_url', type: 'varchar', length: 500 })
  baseUrl: string;

  @Column({ name: 'ws_url', type: 'varchar', length: 500, nullable: true })
  wsUrl: string | null;

  /** AES-256-GCM ciphertext of JSON `{ username, password }`. */
  @Column({ name: 'credentials_encrypted', type: 'text' })
  credentialsEncrypted: string;

  @Column({ name: 'ingest_token', type: 'varchar', length: 100, unique: true })
  ingestToken: string;

  @Column({
    type: 'enum',
    enum: IntegrationStatus,
    default: IntegrationStatus.DISCONNECTED,
  })
  status: IntegrationStatus;

  @Column({ name: 'last_synced_at', type: 'timestamptz', nullable: true })
  lastSyncedAt: Date | null;

  @Column({ name: 'device_linked_count', type: 'integer', default: 0 })
  deviceLinkedCount: number;

  @Column({ name: 'device_created_count', type: 'integer', default: 0 })
  deviceCreatedCount: number;

  @Column({ name: 'last_error', type: 'text', nullable: true })
  lastError: string | null;

  @CreateDateColumn({ name: 'created_at', type: 'timestamptz' })
  createdAt: Date;

  @UpdateDateColumn({ name: 'updated_at', type: 'timestamptz' })
  updatedAt: Date;

  @ManyToOne(() => Organization)
  @JoinColumn({ name: 'organization_id' })
  organization: Organization;
}

Step 3: Commit

git add backend/src/integrations/enums backend/src/integrations/entities/integration-connection.entity.ts
git commit -m "feat(integrations): add IntegrationConnection entity and status enum"

Task 3: `device_integration` entity

Provider-neutral mapping between an HMP device and its remote counterpart.

Files:

Create: backend/src/integrations/entities/device-integration.entity.ts
Step 1: Write the entity

// backend/src/integrations/entities/device-integration.entity.ts
import {
  Column,
  CreateDateColumn,
  Entity,
  Index,
  JoinColumn,
  ManyToOne,
  PrimaryGeneratedColumn,
  UpdateDateColumn,
} from 'typeorm';
import { Device } from '@/devices/entities/device.entity';
import { Organization } from '@/organizations/entities/organization.entity';

@Entity('device_integrations')
@Index('UQ_device_integrations_provider_external', ['provider', 'externalId'], {
  unique: true,
})
@Index('UQ_device_integrations_device_provider', ['deviceId', 'provider'], { unique: true })
export class DeviceIntegration {
  @PrimaryGeneratedColumn('uuid')
  id: string;

  @Column({ name: 'device_id', type: 'uuid' })
  deviceId: string;

  @Column({ name: 'organization_id', type: 'uuid' })
  organizationId: string;

  @Column({ type: 'varchar', length: 50 })
  provider: string;

  @Column({ name: 'external_id', type: 'varchar', length: 100 })
  externalId: string;

  /** AES-256-GCM ciphertext of the remote device access token, when available. */
  @Column({ name: 'credentials_encrypted', type: 'text', nullable: true })
  credentialsEncrypted: string | null;

  @Column({ name: 'last_synced_at', type: 'timestamptz', nullable: true })
  lastSyncedAt: Date | null;

  @CreateDateColumn({ name: 'created_at', type: 'timestamptz' })
  createdAt: Date;

  @UpdateDateColumn({ name: 'updated_at', type: 'timestamptz' })
  updatedAt: Date;

  @ManyToOne(() => Device)
  @JoinColumn({ name: 'device_id' })
  device: Device;

  @ManyToOne(() => Organization)
  @JoinColumn({ name: 'organization_id' })
  organization: Organization;
}

Step 2: Commit

git add backend/src/integrations/entities/device-integration.entity.ts
git commit -m "feat(integrations): add DeviceIntegration mapping entity"

Task 4: Migrations for both tables

Match the existing migration style (QueryRunner + Table/TableIndex, timestamp-prefixed class name).

Files:

Create: backend/src/migrations/1778910000000-CreateIntegrationConnection.ts
Create: backend/src/migrations/1778920000000-CreateDeviceIntegration.ts
Step 1: Write the integration_connections migration

// backend/src/migrations/1778910000000-CreateIntegrationConnection.ts
import type { MigrationInterface, QueryRunner } from 'typeorm';
import { Table, TableIndex } from 'typeorm';

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

  public async up(queryRunner: QueryRunner): Promise<void> {
    await queryRunner.createTable(
      new Table({
        name: 'integration_connections',
        columns: [
          { name: 'id', type: 'uuid', isPrimary: true, default: 'gen_random_uuid()' },
          { name: 'organization_id', type: 'uuid', isNullable: false },
          { name: 'provider', type: 'varchar', length: '50', isNullable: false },
          { name: 'base_url', type: 'varchar', length: '500', isNullable: false },
          { name: 'ws_url', type: 'varchar', length: '500', isNullable: true },
          { name: 'credentials_encrypted', type: 'text', isNullable: false },
          { name: 'ingest_token', type: 'varchar', length: '100', isNullable: false },
          {
            name: 'status',
            type: 'enum',
            enum: ['disconnected', 'connecting', 'syncing', 'connected', 'error'],
            enumName: 'integration_connections_status_enum',
            default: `'disconnected'`,
          },
          { name: 'last_synced_at', type: 'timestamptz', isNullable: true },
          { name: 'device_linked_count', type: 'integer', default: 0 },
          { name: 'device_created_count', type: 'integer', default: 0 },
          { name: 'last_error', type: 'text', isNullable: true },
          { name: 'created_at', type: 'timestamptz', default: 'now()' },
          { name: 'updated_at', type: 'timestamptz', default: 'now()' },
        ],
      }),
      true,
    );

    await queryRunner.createIndex(
      'integration_connections',
      new TableIndex({
        name: 'UQ_integration_connections_org_provider',
        columnNames: ['organization_id', 'provider'],
        isUnique: true,
      }),
    );
    await queryRunner.createIndex(
      'integration_connections',
      new TableIndex({
        name: 'UQ_integration_connections_ingest_token',
        columnNames: ['ingest_token'],
        isUnique: true,
      }),
    );
  }

  public async down(queryRunner: QueryRunner): Promise<void> {
    await queryRunner.dropTable('integration_connections');
    await queryRunner.query('DROP TYPE IF EXISTS "integration_connections_status_enum"');
  }
}

Step 2: Write the device_integrations migration

// backend/src/migrations/1778920000000-CreateDeviceIntegration.ts
import type { MigrationInterface, QueryRunner } from 'typeorm';
import { Table, TableIndex } from 'typeorm';

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

  public async up(queryRunner: QueryRunner): Promise<void> {
    await queryRunner.createTable(
      new Table({
        name: 'device_integrations',
        columns: [
          { name: 'id', type: 'uuid', isPrimary: true, default: 'gen_random_uuid()' },
          { name: 'device_id', type: 'uuid', isNullable: false },
          { name: 'organization_id', type: 'uuid', isNullable: false },
          { name: 'provider', type: 'varchar', length: '50', isNullable: false },
          { name: 'external_id', type: 'varchar', length: '100', isNullable: false },
          { name: 'credentials_encrypted', type: 'text', isNullable: true },
          { name: 'last_synced_at', type: 'timestamptz', isNullable: true },
          { name: 'created_at', type: 'timestamptz', default: 'now()' },
          { name: 'updated_at', type: 'timestamptz', default: 'now()' },
        ],
      }),
      true,
    );

    await queryRunner.createIndex(
      'device_integrations',
      new TableIndex({
        name: 'UQ_device_integrations_provider_external',
        columnNames: ['provider', 'external_id'],
        isUnique: true,
      }),
    );
    await queryRunner.createIndex(
      'device_integrations',
      new TableIndex({
        name: 'UQ_device_integrations_device_provider',
        columnNames: ['device_id', 'provider'],
        isUnique: true,
      }),
    );
  }

  public async down(queryRunner: QueryRunner): Promise<void> {
    await queryRunner.dropTable('device_integrations');
  }
}

Step 3: Run migrations against the dev DB to verify they apply

Run: cd backend && bun run migration:run
Expected: both migrations execute without error; integration_connections and device_integrations tables exist.

Step 4: Verify rollback works, then re-apply

Run: cd backend && bun run migration:revert && bun run migration:revert && bun run migration:run
Expected: reverts drop both tables cleanly, re-run recreates them. (Each migration:revert undoes one migration.)

Step 5: Commit

git add backend/src/migrations/1778910000000-CreateIntegrationConnection.ts backend/src/migrations/1778920000000-CreateDeviceIntegration.ts
git commit -m "feat(integrations): add migrations for connection and device-mapping tables"

Task 5: `DeviceConnector` interface + shared types

Files:

Create: backend/src/integrations/connectors/device-connector.interface.ts
Step 1: Write the interface and types

// backend/src/integrations/connectors/device-connector.interface.ts

/** Runtime connection settings a connector needs, already decrypted. */
export interface ConnectorConfig {
  baseUrl: string;
  username: string;
  password: string;
  wsUrl?: string | null;
}

export interface RemoteDevice {
  externalId: string;
  name: string;
}

export interface RemoteDeviceRef {
  externalId: string;
}

export interface RemoteCredentials {
  accessToken: string;
}

export interface ConnectorRpcRequest {
  method: string;
  params: unknown;
}

/**
 * Provider-neutral device connector. Implementations wrap one external IoT
 * platform (ThingsBoard now; MQTT later). Instances are config-bound: one per
 * organization connection, created by the registry.
 */
export interface DeviceConnector {
  readonly provider: string;

  /** Authenticate against the provider; throws on failure. */
  verifyConnection(): Promise<void>;

  /** List devices that exist on the remote provider. */
  discoverDevices(): Promise<RemoteDevice[]>;

  /** Create a device on the remote provider. */
  pushDevice(input: { name: string }): Promise<RemoteDeviceRef>;

  /** Delete a device on the remote provider. */
  deleteRemoteDevice(externalId: string): Promise<void>;

  /** Fetch the remote device's access credentials. */
  getRemoteDeviceCredentials(externalId: string): Promise<RemoteCredentials>;

  /** Send an RPC command to a remote device. */
  sendRpc(externalId: string, rpc: ConnectorRpcRequest, oneway?: boolean): Promise<unknown>;
}

Step 2: Commit

git add backend/src/integrations/connectors/device-connector.interface.ts
git commit -m "feat(integrations): add DeviceConnector interface and shared types"

Task 6: `ConnectorRegistry`

Maps a provider string to a factory that builds a config-bound DeviceConnector.

Files:

Create: backend/src/integrations/connectors/connector-registry.ts
Test: backend/src/integrations/connectors/connector-registry.spec.ts
Step 1: Write the failing test

// backend/src/integrations/connectors/connector-registry.spec.ts
import type {
  ConnectorConfig,
  DeviceConnector,
} from '@/integrations/connectors/device-connector.interface';
import { ConnectorRegistry } from '@/integrations/connectors/connector-registry';

function fakeConnector(provider: string): DeviceConnector {
  return {
    provider,
    verifyConnection: jest.fn(),
    discoverDevices: jest.fn(),
    pushDevice: jest.fn(),
    deleteRemoteDevice: jest.fn(),
    getRemoteDeviceCredentials: jest.fn(),
    sendRpc: jest.fn(),
  } as unknown as DeviceConnector;
}

describe('ConnectorRegistry', () => {
  let registry: ConnectorRegistry;
  const config: ConnectorConfig = { baseUrl: 'http://tb', username: 'u', password: 'p' };

  beforeEach(() => {
    registry = new ConnectorRegistry();
  });

  it('creates a connector for a registered provider', () => {
    registry.register('thingsboard', (cfg) => fakeConnector(`thingsboard@${cfg.baseUrl}`));
    const connector = registry.create('thingsboard', config);
    expect(connector.provider).toBe('thingsboard@http://tb');
  });

  it('reports whether a provider is registered', () => {
    expect(registry.has('thingsboard')).toBe(false);
    registry.register('thingsboard', () => fakeConnector('thingsboard'));
    expect(registry.has('thingsboard')).toBe(true);
  });

  it('throws when creating an unregistered provider', () => {
    expect(() => registry.create('mqtt', config)).toThrow(/no connector registered.*mqtt/i);
  });

  it('throws when registering the same provider twice', () => {
    registry.register('thingsboard', () => fakeConnector('thingsboard'));
    expect(() => registry.register('thingsboard', () => fakeConnector('thingsboard'))).toThrow(
      /already registered/i,
    );
  });
});

Step 2: Run test to verify it fails

Run: make backend-test FILE=backend/src/integrations/connectors/connector-registry.spec.ts
Expected: FAIL - cannot find module connector-registry.

Step 3: Write minimal implementation

// backend/src/integrations/connectors/connector-registry.ts
import { Injectable } from '@nestjs/common';
import type {
  ConnectorConfig,
  DeviceConnector,
} from '@/integrations/connectors/device-connector.interface';

export type ConnectorFactory = (config: ConnectorConfig) => DeviceConnector;

@Injectable()
export class ConnectorRegistry {
  private readonly factories = new Map<string, ConnectorFactory>();

  register(provider: string, factory: ConnectorFactory): void {
    if (this.factories.has(provider)) {
      throw new Error(`Connector for provider "${provider}" is already registered.`);
    }
    this.factories.set(provider, factory);
  }

  has(provider: string): boolean {
    return this.factories.has(provider);
  }

  create(provider: string, config: ConnectorConfig): DeviceConnector {
    const factory = this.factories.get(provider);
    if (!factory) {
      throw new Error(`No connector registered for provider "${provider}".`);
    }
    return factory(config);
  }
}

Step 4: Run test to verify it passes

Run: make backend-test FILE=backend/src/integrations/connectors/connector-registry.spec.ts
Expected: PASS (4 tests).

Step 5: Commit

git add backend/src/integrations/connectors/connector-registry.ts backend/src/integrations/connectors/connector-registry.spec.ts
git commit -m "feat(integrations): add ConnectorRegistry"

Task 7: ThingsBoard connector implementation

Port the REST logic from backend/src/thingsboard/thingsboard-client.service.ts into a config-bound DeviceConnector. Keep the ThingsBoardApiError semantics. The connector takes a ConnectorConfig in its constructor (no ConfigService, no env vars). discoverDevices uses TB’s tenant device page endpoint.

Files:

Create: backend/src/integrations/connectors/thingsboard/thingsboard.connector.ts
Test: backend/src/integrations/connectors/thingsboard/thingsboard.connector.spec.ts
Step 1: Write the failing test

Mocks global fetch. Verifies login-then-request flow, device discovery paging shape, and RPC.

// backend/src/integrations/connectors/thingsboard/thingsboard.connector.spec.ts
import { ThingsBoardConnector } from '@/integrations/connectors/thingsboard/thingsboard.connector';

function jsonResponse(body: unknown, status = 200): Response {
  return {
    ok: status >= 200 && status < 300,
    status,
    text: async () => JSON.stringify(body),
  } as unknown as Response;
}

describe('ThingsBoardConnector', () => {
  const config = {
    baseUrl: 'http://tb:8090/',
    username: 'tenant@thingsboard.org',
    password: 'tenant',
  };
  let connector: ThingsBoardConnector;
  let fetchMock: jest.Mock;

  beforeEach(() => {
    connector = new ThingsBoardConnector(config);
    fetchMock = jest.fn();
    global.fetch = fetchMock as unknown as typeof fetch;
  });

  it('verifyConnection logs in via /api/auth/login', async () => {
    fetchMock.mockResolvedValueOnce(jsonResponse({ token: 'jwt', refreshToken: 'r' }));
    await connector.verifyConnection();
    expect(fetchMock).toHaveBeenCalledWith(
      'http://tb:8090/api/auth/login',
      expect.objectContaining({ method: 'POST' }),
    );
  });

  it('verifyConnection throws when login fails', async () => {
    fetchMock.mockResolvedValueOnce(jsonResponse({}, 401));
    await expect(connector.verifyConnection()).rejects.toThrow(/login failed/i);
  });

  it('discoverDevices returns externalId + name from the tenant device page', async () => {
    fetchMock
      .mockResolvedValueOnce(jsonResponse({ token: 'jwt', refreshToken: 'r' }))
      .mockResolvedValueOnce(
        jsonResponse({
          data: [{ id: { id: 'tb-uuid-1' }, name: 'SN-001' }],
          hasNext: false,
        }),
      );
    const devices = await connector.discoverDevices();
    expect(devices).toEqual([{ externalId: 'tb-uuid-1', name: 'SN-001' }]);
  });

  it('sendRpc posts to the twoway RPC endpoint', async () => {
    fetchMock
      .mockResolvedValueOnce(jsonResponse({ token: 'jwt', refreshToken: 'r' }))
      .mockResolvedValueOnce(jsonResponse({ result: 'ok' }));
    const result = await connector.sendRpc('tb-uuid-1', { method: 'setState', params: 1 });
    expect(fetchMock).toHaveBeenLastCalledWith(
      'http://tb:8090/api/rpc/twoway/tb-uuid-1',
      expect.objectContaining({ method: 'POST' }),
    );
    expect(result).toEqual({ result: 'ok' });
  });
});

Step 2: Run test to verify it fails

Run: make backend-test FILE=backend/src/integrations/connectors/thingsboard/thingsboard.connector.spec.ts
Expected: FAIL - cannot find module thingsboard.connector.

Step 3: Write the implementation

Ports the existing client, adapted to ConnectorConfig and the DeviceConnector interface. discoverDevices calls TB’s /api/tenant/devices paged endpoint.

// backend/src/integrations/connectors/thingsboard/thingsboard.connector.ts
import { Logger } from '@nestjs/common';
import type {
  ConnectorConfig,
  ConnectorRpcRequest,
  DeviceConnector,
  RemoteCredentials,
  RemoteDevice,
  RemoteDeviceRef,
} from '@/integrations/connectors/device-connector.interface';

interface TbLoginResponse {
  token: string;
  refreshToken: string;
}
interface TbEntityId {
  id: string;
  entityType?: string;
}
interface TbDeviceResponse {
  id: TbEntityId;
  name: string;
}
interface TbDevicePage {
  data: TbDeviceResponse[];
  hasNext: boolean;
}
interface TbCredentialsResponse {
  credentialsId: string;
  credentialsType: string;
}
interface TbErrorBody {
  message?: string;
  errorCode?: number;
}

/** Thrown for any non-2xx ThingsBoard REST response. */
export class ThingsBoardApiError extends Error {
  constructor(
    message: string,
    public readonly status: number,
    public readonly tbErrorCode?: number,
  ) {
    super(message);
    this.name = 'ThingsBoardApiError';
  }
}

/**
 * ThingsBoard (Community Edition) implementation of DeviceConnector.
 * Config-bound: one instance per organization connection. Ported from the
 * former env-var-driven ThingsBoardClientService.
 */
export class ThingsBoardConnector implements DeviceConnector {
  readonly provider = 'thingsboard';
  private readonly logger = new Logger(ThingsBoardConnector.name);
  private readonly baseUrl: string;
  private readonly username: string;
  private readonly password: string;

  private token: string | null = null;
  private loginInFlight: Promise<string> | null = null;

  constructor(config: ConnectorConfig) {
    this.baseUrl = config.baseUrl.replace(/\/$/, '');
    this.username = config.username;
    this.password = config.password;
  }

  async verifyConnection(): Promise<void> {
    this.token = null;
    await this.login();
  }

  async discoverDevices(): Promise<RemoteDevice[]> {
    const devices: RemoteDevice[] = [];
    const pageSize = 100;
    let page = 0;
    let hasNext = true;
    while (hasNext) {
      const result = await this.request<TbDevicePage>(
        `/api/tenant/devices?pageSize=${pageSize}&page=${page}`,
      );
      for (const device of result.data) {
        devices.push({ externalId: device.id.id, name: device.name });
      }
      hasNext = result.hasNext;
      page += 1;
    }
    return devices;
  }

  async pushDevice(input: { name: string }): Promise<RemoteDeviceRef> {
    const device = await this.request<TbDeviceResponse>('/api/device', {
      method: 'POST',
      body: JSON.stringify({ name: input.name }),
    });
    return { externalId: device.id.id };
  }

  async deleteRemoteDevice(externalId: string): Promise<void> {
    await this.request<void>(`/api/device/${externalId}`, { method: 'DELETE' });
  }

  async getRemoteDeviceCredentials(externalId: string): Promise<RemoteCredentials> {
    const credentials = await this.request<TbCredentialsResponse>(
      `/api/device/${externalId}/credentials`,
    );
    return { accessToken: credentials.credentialsId };
  }

  async sendRpc(
    externalId: string,
    rpc: ConnectorRpcRequest,
    oneway = false,
  ): Promise<unknown> {
    const direction = oneway ? 'oneway' : 'twoway';
    return this.request<unknown>(`/api/rpc/${direction}/${externalId}`, {
      method: 'POST',
      body: JSON.stringify(rpc),
    });
  }

  private async login(): Promise<string> {
    if (this.loginInFlight) {
      return this.loginInFlight;
    }
    this.loginInFlight = (async () => {
      const response = await fetch(`${this.baseUrl}/api/auth/login`, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ username: this.username, password: this.password }),
      });
      if (!response.ok) {
        throw new Error(`ThingsBoard login failed with status ${response.status}`);
      }
      const data = JSON.parse(await response.text()) as TbLoginResponse;
      this.token = data.token;
      return data.token;
    })();
    try {
      return await this.loginInFlight;
    } finally {
      this.loginInFlight = null;
    }
  }

  private async request<T>(path: string, init: RequestInit = {}, retryOn401 = true): Promise<T> {
    const token = this.token ?? (await this.login());
    const response = await fetch(`${this.baseUrl}${path}`, {
      ...init,
      headers: {
        'Content-Type': 'application/json',
        'X-Authorization': `Bearer ${token}`,
        ...(init.headers ?? {}),
      },
    });

    if (response.status === 401 && retryOn401) {
      this.token = null;
      await this.login();
      return this.request<T>(path, init, false);
    }

    if (!response.ok) {
      const body = await response.text();
      let tbErrorCode: number | undefined;
      let detail = body;
      try {
        const parsed = JSON.parse(body) as TbErrorBody;
        tbErrorCode = parsed.errorCode;
        detail = parsed.message ?? body;
      } catch {
        // Body wasn't TB's structured JSON error — keep the raw text.
      }
      throw new ThingsBoardApiError(
        `ThingsBoard request ${path} failed (${response.status}): ${detail}`,
        response.status,
        tbErrorCode,
      );
    }

    const text = await response.text();
    return (text ? JSON.parse(text) : undefined) as T;
  }
}

Step 4: Run test to verify it passes

Run: make backend-test FILE=backend/src/integrations/connectors/thingsboard/thingsboard.connector.spec.ts
Expected: PASS (4 tests).

Step 5: Commit

git add backend/src/integrations/connectors/thingsboard
git commit -m "feat(integrations): add ThingsBoard DeviceConnector implementation"

Task 8: `ConnectorConnectionManager`

Resolves a config-bound connector for an organization: loads the integration_connection, decrypts credentials, builds via the registry, and caches per org. Cache is invalidated by clearCache.

Files:

Create: backend/src/integrations/connector-connection.manager.ts
Test: backend/src/integrations/connector-connection.manager.spec.ts
Step 1: Write the failing test

// backend/src/integrations/connector-connection.manager.spec.ts
import { NotFoundException } from '@nestjs/common';
import type { Repository } from 'typeorm';
import { ConnectorRegistry } from '@/integrations/connectors/connector-registry';
import type { DeviceConnector } from '@/integrations/connectors/device-connector.interface';
import { ConnectorConnectionManager } from '@/integrations/connector-connection.manager';
import { CredentialCipher } from '@/integrations/crypto/credential-cipher';
import { IntegrationConnection } from '@/integrations/entities/integration-connection.entity';
import { IntegrationStatus } from '@/integrations/enums/integration-status.enum';

describe('ConnectorConnectionManager', () => {
  const key = Buffer.alloc(32, 9).toString('base64');
  let cipher: CredentialCipher;
  let registry: ConnectorRegistry;
  let repo: { findOne: jest.Mock };
  let manager: ConnectorConnectionManager;
  let created: number;

  beforeEach(() => {
    cipher = new CredentialCipher(key);
    registry = new ConnectorRegistry();
    created = 0;
    registry.register('thingsboard', (cfg) => {
      created += 1;
      return { provider: 'thingsboard', baseUrl: cfg.baseUrl } as unknown as DeviceConnector;
    });
    repo = { findOne: jest.fn() };
    manager = new ConnectorConnectionManager(
      repo as unknown as Repository<IntegrationConnection>,
      registry,
      cipher,
    );
  });

  function connectionRow(): IntegrationConnection {
    return {
      id: 'c1',
      organizationId: 'org-1',
      provider: 'thingsboard',
      baseUrl: 'http://tb',
      wsUrl: null,
      credentialsEncrypted: cipher.encrypt(
        JSON.stringify({ username: 'u@tb', password: 'pw' }),
      ),
      status: IntegrationStatus.CONNECTED,
    } as IntegrationConnection;
  }

  it('builds a connector from the decrypted org connection', async () => {
    repo.findOne.mockResolvedValue(connectionRow());
    const connector = await manager.getConnector('org-1');
    expect(connector.provider).toBe('thingsboard');
  });

  it('caches the connector per organization', async () => {
    repo.findOne.mockResolvedValue(connectionRow());
    await manager.getConnector('org-1');
    await manager.getConnector('org-1');
    expect(created).toBe(1);
    expect(repo.findOne).toHaveBeenCalledTimes(1);
  });

  it('rebuilds after clearCache', async () => {
    repo.findOne.mockResolvedValue(connectionRow());
    await manager.getConnector('org-1');
    manager.clearCache('org-1');
    await manager.getConnector('org-1');
    expect(created).toBe(2);
  });

  it('throws NotFound when the org has no connection', async () => {
    repo.findOne.mockResolvedValue(null);
    await expect(manager.getConnector('org-1')).rejects.toBeInstanceOf(NotFoundException);
  });
});

Step 2: Run test to verify it fails

Run: make backend-test FILE=backend/src/integrations/connector-connection.manager.spec.ts
Expected: FAIL - cannot find module connector-connection.manager.

Step 3: Write the implementation

// backend/src/integrations/connector-connection.manager.ts
import { Injectable, NotFoundException } from '@nestjs/common';
import { InjectRepository } from '@nestjs/typeorm';
import { Repository } from 'typeorm';
import { ConnectorRegistry } from '@/integrations/connectors/connector-registry';
import type { DeviceConnector } from '@/integrations/connectors/device-connector.interface';
import { CredentialCipher } from '@/integrations/crypto/credential-cipher';
import { IntegrationConnection } from '@/integrations/entities/integration-connection.entity';

interface StoredCredentials {
  username: string;
  password: string;
}

const DEFAULT_PROVIDER = 'thingsboard';

/**
 * Resolves a config-bound DeviceConnector for an organization. Loads the org's
 * IntegrationConnection, decrypts its credentials, and builds (and caches) a
 * connector via the registry. Call clearCache when the connection config
 * changes.
 */
@Injectable()
export class ConnectorConnectionManager {
  private readonly cache = new Map<string, DeviceConnector>();

  constructor(
    @InjectRepository(IntegrationConnection)
    private readonly connectionRepo: Repository<IntegrationConnection>,
    private readonly registry: ConnectorRegistry,
    private readonly cipher: CredentialCipher,
  ) {}

  async getConnector(
    organizationId: string,
    provider: string = DEFAULT_PROVIDER,
  ): Promise<DeviceConnector> {
    const cacheKey = `${organizationId}:${provider}`;
    const cached = this.cache.get(cacheKey);
    if (cached) {
      return cached;
    }

    const connection = await this.connectionRepo.findOne({
      where: { organizationId, provider },
    });
    if (!connection) {
      throw new NotFoundException(
        `No ${provider} integration configured for organization ${organizationId}.`,
      );
    }

    const credentials = JSON.parse(
      this.cipher.decrypt(connection.credentialsEncrypted),
    ) as StoredCredentials;

    const connector = this.registry.create(provider, {
      baseUrl: connection.baseUrl,
      username: credentials.username,
      password: credentials.password,
      wsUrl: connection.wsUrl,
    });

    this.cache.set(cacheKey, connector);
    return connector;
  }

  clearCache(organizationId: string, provider: string = DEFAULT_PROVIDER): void {
    this.cache.delete(`${organizationId}:${provider}`);
  }
}

Step 4: Run test to verify it passes

Run: make backend-test FILE=backend/src/integrations/connector-connection.manager.spec.ts
Expected: PASS (4 tests).

Step 5: Commit

git add backend/src/integrations/connector-connection.manager.ts backend/src/integrations/connector-connection.manager.spec.ts
git commit -m "feat(integrations): add ConnectorConnectionManager with per-org caching"

Task 9: `IntegrationsModule` wiring + entity registration

Wire the module: provide the cipher (built from env), the registry, and the connection manager; register the TB factory on module init; register entities with TypeORM.

Files:

Create: backend/src/integrations/integrations.module.ts
Modify: backend/src/app.module.ts:90-132 (entities list) and imports section (line ~160).
Step 1: Write the module

// backend/src/integrations/integrations.module.ts
import { Module, type OnModuleInit } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { TypeOrmModule } from '@nestjs/typeorm';
import { ConnectorConnectionManager } from '@/integrations/connector-connection.manager';
import { ConnectorRegistry } from '@/integrations/connectors/connector-registry';
import { ThingsBoardConnector } from '@/integrations/connectors/thingsboard/thingsboard.connector';
import { CredentialCipher } from '@/integrations/crypto/credential-cipher';
import { DeviceIntegration } from '@/integrations/entities/device-integration.entity';
import { IntegrationConnection } from '@/integrations/entities/integration-connection.entity';

@Module({
  imports: [TypeOrmModule.forFeature([IntegrationConnection, DeviceIntegration])],
  providers: [
    {
      provide: CredentialCipher,
      useFactory: (config: ConfigService) => {
        const key = config.get<string>('INTEGRATION_ENCRYPTION_KEY');
        if (!key) {
          throw new Error('INTEGRATION_ENCRYPTION_KEY is not configured.');
        }
        return new CredentialCipher(key);
      },
      inject: [ConfigService],
    },
    ConnectorRegistry,
    ConnectorConnectionManager,
  ],
  exports: [ConnectorConnectionManager, ConnectorRegistry, CredentialCipher, TypeOrmModule],
})
export class IntegrationsModule implements OnModuleInit {
  constructor(private readonly registry: ConnectorRegistry) {}

  onModuleInit(): void {
    this.registry.register('thingsboard', (config) => new ThingsBoardConnector(config));
  }
}

Step 2: Register entities in app.module.ts

In the entities: [ ... ] array (currently ending at MonitoringCommandDispatchRecord, on line 131), add the two new entities and their imports at the top of the file:

import { IntegrationConnection } from '@/integrations/entities/integration-connection.entity';
import { DeviceIntegration } from '@/integrations/entities/device-integration.entity';

Add to the array after MonitoringCommandDispatchRecord,:

            MonitoringCommandDispatchRecord,
            IntegrationConnection,
            DeviceIntegration,

Add IntegrationsModule to the module imports list (after MonitoringModule, on line 160):

import { IntegrationsModule } from '@/integrations/integrations.module';
// ...
    MonitoringModule,
    IntegrationsModule,

Step 3: Verify the app boots and the whole integrations suite passes

Run: cd backend && bun run build
Expected: TypeScript compiles with no errors.

Run: make backend-test FILE=backend/src/integrations
Expected: all integrations specs PASS (cipher, registry, connector, manager).

Step 4: Set INTEGRATION_ENCRYPTION_KEY in the local .env

The module throws on boot without it. Generate and add to backend/.env:

Run: node -e "console.log('INTEGRATION_ENCRYPTION_KEY=' + require('crypto').randomBytes(32).toString('base64'))"
Then paste the printed line into backend/.env.

Step 5: Verify app starts

Run: cd backend && bun run start:dev (let it boot, confirm no DI/crypto errors, then stop)
Expected: Nest boots; IntegrationsModule initializes; log shows no INTEGRATION_ENCRYPTION_KEY error.

Step 6: Commit

git add backend/src/integrations/integrations.module.ts backend/src/app.module.ts
git commit -m "feat(integrations): wire IntegrationsModule and register entities"

Task 10: Lint the touched workspace

Step 1: Run Biome

Run: bun run check
Expected: no lint/format errors in backend/src/integrations/. Fix any reported issues.

Step 2: Commit if Biome made changes

git add -A
git commit -m "chore(integrations): biome formatting"

Self-Review Notes (author)

Spec coverage (this plan): Section 3.1 (interface, registry, connection manager) - Tasks 5, 6, 8. Section 3.2 (TB impl, ported REST logic, per-org config) - Task 7. Section 4.1 (integration_connection) - Tasks 2, 4. Section 4.2 (device_integration) - Tasks 3, 4. Section 4.3 (encryption) - Task 1. Deferred to later plans (correctly out of scope here): 4.4 cleanup + column drop (Plan 3), Sections 5-11 (Plans 2-3).
Type consistency: ConnectorConfig, RemoteDevice, RemoteDeviceRef, RemoteCredentials, ConnectorRpcRequest, DeviceConnector defined once in Task 5 and used verbatim in Tasks 6, 7, 8. IntegrationStatus defined in Task 2, used in Tasks 4, 8. ConnectorFactory defined in Task 6, used in Task 9.
No consumer changes: the existing Device.thingsboardDeviceId/accessToken columns and ThingsboardModule remain untouched; the migration to device_integration and column drop happen in Plan 3 after consumers are switched over.
Migration timestamps (1778910000000, 1778920000000) are after the latest existing (1778900000000); confirm ordering still holds at execution time.

ThingsBoard Connector Plugin - Foundation (Plan 1 of 3) Implementation Plan

File Structure

Task 1: Credential encryption utility

Task 2: Integration status enum + integration_connection entity

Task 3: device_integration entity

Task 4: Migrations for both tables

Task 5: DeviceConnector interface + shared types

Task 6: ConnectorRegistry

Task 7: ThingsBoard connector implementation

Task 8: ConnectorConnectionManager

Task 9: IntegrationsModule wiring + entity registration

Task 10: Lint the touched workspace

Self-Review Notes (author)

onThisPage

Task 2: Integration status enum + `integration_connection` entity

Task 3: `device_integration` entity

Task 5: `DeviceConnector` interface + shared types

Task 6: `ConnectorRegistry`

Task 8: `ConnectorConnectionManager`

Task 9: `IntegrationsModule` wiring + entity registration