trm/tcp-ingestion
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement Phase 1 tasks 1.1-1.4 (scaffold + core shell + Teltonika framing)

Browse Source 
- Project scaffold (Node 22 + TS 5 + pnpm + vitest + ESLint flat config)
- Core shell: TCP server, session loop, adapter registry, types
- Configuration (zod-validated env) and pino logger
- Teltonika adapter: IMEI handshake, frame envelope, CRC-16/IBM,
  codec dispatch registry, DeviceAuthority seam (AllowAllAuthority default)

Codec data parsers (1.5-1.7), Redis publisher (1.8), and downstream
tasks remain. 36 tests covering CRC, framing, handshake, device
authority, config, and core server. typecheck/lint/test/build all clean.
This commit is contained in:
Julian Çuni
2026-04-30 15:47:26 +02:00
parent c8a5f4cd68
commit 1e9219d14a
35 changed files with 6217 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.dockerignore
+11
View File
@@ -0,0 +1,11 @@
node_modules/
dist/
coverage/
.env
.env.local
*.log
.git/
.planning/
test/
*.md
!README.md
.gitignore
+6
View File
@@ -0,0 +1,6 @@
node_modules/
dist/
coverage/
.env
.env.local
*.log
.prettierrc
+7
View File
@@ -0,0 +1,7 @@
{
"semi": true,
"singleQuote": true,
"printWidth": 100,
"trailingComma": "all",
"tabWidth": 2
}
eslint.config.js
+71
View File
@@ -0,0 +1,71 @@
// @ts-check
import tseslint from "@typescript-eslint/eslint-plugin";
import tsParser from "@typescript-eslint/parser";
import importPlugin from "eslint-plugin-import";
import { fileURLToPath } from "node:url";
import { dirname, join } from "node:path";
const __filename = fileURLToPath(import.meta.url);
const __dirname = dirname(__filename);
/** @type {import('eslint').Linter.Config[]} */
export default [
{
ignores: ["dist/**", "node_modules/**", "coverage/**"],
},
{
files: ["src/**/*.ts", "test/**/*.ts"],
plugins: {
"@typescript-eslint": tseslint,
import: importPlugin,
},
languageOptions: {
parser: tsParser,
parserOptions: {
project: "./tsconfig.test.json",
tsconfigRootDir: __dirname,
ecmaVersion: 2022,
sourceType: "module",
},
},
settings: {
"import/resolver": {
typescript: {
project: join(__dirname, "tsconfig.test.json"),
},
},
},
rules: {
// TypeScript strict promise rules — critical in a TCP server
"@typescript-eslint/no-floating-promises": "error",
"@typescript-eslint/no-misused-promises": "error",
// General quality
"@typescript-eslint/no-explicit-any": "error",
"@typescript-eslint/no-unused-vars": [
"error",
{ argsIgnorePattern: "^_", varsIgnorePattern: "^_" },
],
"@typescript-eslint/consistent-type-imports": [
"error",
{ prefer: "type-imports" },
],
// Adapter isolation: core/ must NEVER import from adapters/
"import/no-restricted-paths": [
"error",
{
basePath: __dirname,
zones: [
{
target: "src/core",
from: "src/adapters",
message:
"src/core must not import from src/adapters — adapters depend on core, not the reverse.",
},
],
},
],
},
},
];
package.json
+39
View File
@@ -0,0 +1,39 @@
{
"name": "tcp-ingestion",
"version": "0.1.0",
"description": "TCP server ingesting Teltonika GPS telemetry and publishing normalized Position records to Redis Streams",
"type": "module",
"engines": {
"node": ">=22"
},
"scripts": {
"build": "tsc --project tsconfig.json",
"dev": "tsx watch src/main.ts",
"start": "node dist/main.js",
"test": "vitest run",
"test:watch": "vitest",
"lint": "eslint .",
"format": "prettier --write .",
"typecheck": "tsc --noEmit"
},
"dependencies": {
"ioredis": "^5.3.2",
"pino": "^9.5.0",
"prom-client": "^15.1.3",
"zod": "^3.23.8"
},
"devDependencies": {
"@types/node": "^22.10.0",
"@typescript-eslint/eslint-plugin": "^8.19.0",
"@typescript-eslint/parser": "^8.19.0",
"@vitest/coverage-v8": "^2.1.8",
"eslint": "^9.17.0",
"eslint-import-resolver-typescript": "^4.4.4",
"eslint-plugin-import": "^2.31.0",
"pino-pretty": "^13.0.0",
"prettier": "^3.4.2",
"tsx": "^4.19.2",
"typescript": "^5.7.2",
"vitest": "^2.1.8"
}
}
pnpm-lock.yaml
Generated
+4259
View File
File diff suppressed because it is too large Load Diff
src/adapters/teltonika/codec/command/.gitkeep
View File
src/adapters/teltonika/codec/data/.gitkeep
View File
src/adapters/teltonika/codec/registry.ts
+53
View File
@@ -0,0 +1,53 @@
import type { Logger } from 'pino';
import type { Position } from '../../../core/types.js';
/**
* Context passed to codec data handlers. Narrow contract: parsers receive
* only what they need (IMEI for device_id, publish for emitting records,
* logger for structured output).
*
* Phase 2 will add a `respond(bytes: Buffer) => void` to this ctx for command
* codecs that write back to the socket. Data codecs (8, 8E, 16) never write.
*/
export type CodecHandlerContext = {
readonly imei: string;
readonly publish: (position: Position) => Promise<void>;
readonly logger: Logger;
};
/**
* A codec data handler registered in the flat dispatch registry.
* Each handler owns one codec ID and is responsible for:
* - Parsing N1/N2 count validation (within the body it receives)
* - Producing Position records via ctx.publish
* - Returning the number of records accepted (used for the TCP ACK)
*/
export interface CodecDataHandler {
readonly codec_id: number;
handle(
body: Buffer, // Full body: CodecID (1B) + N1 (1B) + records + N2 (1B)
ctx: CodecHandlerContext,
): Promise<{ recordCount: number }>;
}
/**
* Flat registry mapping codec IDs to their handlers.
* Phase 1 registers 0x08, 0x8E, 0x10. Phase 2 will register 0x0C, 0x0E
* alongside without modifying Phase 1 paths.
*/
export class CodecRegistry {
private readonly handlers = new Map<number, CodecDataHandler>();
register(handler: CodecDataHandler): void {
if (this.handlers.has(handler.codec_id)) {
throw new Error(
`Codec 0x${handler.codec_id.toString(16).toUpperCase()} is already registered.`,
);
}
this.handlers.set(handler.codec_id, handler);
}
get(codecId: number): CodecDataHandler | undefined {
return this.handlers.get(codecId);
}
}
src/adapters/teltonika/crc.ts
+45
View File
@@ -0,0 +1,45 @@
/**
* CRC-16/IBM (also known as CRC-16/ARC).
* Polynomial: 0xA001 (reflected form of 0x8005).
* Initial value: 0x0000. No final XOR.
*
* Teltonika uses this to protect the AVL data body from CodecID through N2
* inclusive. The 4-byte CRC field in the frame carries the result in the lower
* 2 bytes; upper 2 bytes are always zero.
*
* Implementation uses a precomputed 256-entry lookup table for performance —
* protocol parsing is on the hot path and frame bodies can be up to ~1280 bytes.
*/
// Build lookup table at module load time (one-off cost, not per-frame)
const CRC_TABLE: Uint16Array = buildTable();
function buildTable(): Uint16Array {
const table = new Uint16Array(256);
for (let i = 0; i < 256; i++) {
let crc = i;
for (let j = 0; j < 8; j++) {
if ((crc & 0x0001) !== 0) {
crc = (crc >>> 1) ^ 0xa001;
} else {
crc = crc >>> 1;
}
}
table[i] = crc;
}
return table;
}
/**
* Computes CRC-16/IBM over the given buffer.
* Returns a 16-bit unsigned integer in the range [0x0000, 0xFFFF].
*/
export function crc16Ibm(buf: Buffer): number {
let crc = 0x0000;
for (let i = 0; i < buf.length; i++) {
const byte = buf[i] ?? 0;
const tableIndex = (crc ^ byte) & 0xff;
crc = (crc >>> 8) ^ (CRC_TABLE[tableIndex] ?? 0);
}
return crc & 0xffff;
}
src/adapters/teltonika/device-authority.ts
+25
View File
@@ -0,0 +1,25 @@
/**
* Determines whether a connecting device is a recognized member of the fleet.
*
* Phase 1 default: AllowAllAuthority — every IMEI returns 'known'. This keeps
* all current devices working while making the label available for metrics.
*
* Phase 1.13 will add RedisAllowListAuthority, which reads a Directus-published
* allow-list from Redis and returns 'unknown' for devices not in the list.
*
* The STRICT_DEVICE_AUTH flag causes 'unknown' devices to be rejected (0x00
* handshake response + socket destroy). It is off by default.
*/
export interface DeviceAuthority {
check(imei: string): Promise<'known' | 'unknown'>;
}
/**
* Permissive default authority — always returns 'known'.
* Used until a real authority is configured (task 1.13).
*/
export class AllowAllAuthority implements DeviceAuthority {
async check(_imei: string): Promise<'known'> {
return 'known';
}
}
src/adapters/teltonika/frame.ts
+209
View File
@@ -0,0 +1,209 @@
import type * as net from 'node:net';
import { crc16Ibm } from './crc.js';
/**
* Maximum permitted DataFieldLength value. Covers the 1280-byte max AVL packet
* for most devices, with headroom. Frames above this threshold are structurally
* malformed — we drop the connection rather than allocating unbounded memory.
*/
export const MAX_AVL_PACKET_SIZE = 1300;
/**
* Reason a frame read attempt failed in a way that should close the connection.
*/
export type FrameDropReason =
| 'invalid_preamble'
| 'oversized_frame'
| 'n1_n2_mismatch'
| 'socket_closed';
export class FrameDropError extends Error {
constructor(
public readonly reason: FrameDropReason,
message: string,
) {
super(message);
this.name = 'FrameDropError';
}
}
/**
* Parsed AVL frame envelope. CRC validity is surfaced here so the session loop
* can decide whether to ACK without the frame handler needing to re-compute.
*/
export type Frame = {
readonly codecId: number;
readonly payload: Buffer; // full body: CodecID + N1 + records + N2
readonly crcValid: boolean;
readonly expectedCrc: number;
readonly computedCrc: number;
};
// ---------------------------------------------------------------------------
// BufferedReader
// ---------------------------------------------------------------------------
type PendingRead = {
needed: number;
resolve: (buf: Buffer) => void;
reject: (err: Error) => void;
};
/**
* Accumulates bytes arriving across multiple `data` events and satisfies
* exact-byte-count read requests. This handles the reality that TCP is a stream
* — a 100-byte read may arrive as five 20-byte chunks.
*
* Design: a single pending read at a time. The session loop is sequential
* (each `await reader.readExact(n)` completes before the next starts) so
* concurrent reads are not needed.
*/
export class BufferedReader {
private readonly chunks: Buffer[] = [];
private buffered = 0;
private pending: PendingRead | null = null;
private closed = false;
private closeError: Error | null = null;
constructor(private readonly socket: net.Socket) {
socket.on('data', (chunk: Buffer) => this.onData(chunk));
socket.on('close', () => this.onClose());
socket.on('error', (err) => this.onSocketError(err));
}
/**
* Reads exactly `n` bytes. Returns a Promise that resolves once the bytes are
* available. Rejects if the socket closes before the read completes.
*/
readExact(n: number): Promise<Buffer> {
// Fast path: data already buffered
if (this.buffered >= n) {
return Promise.resolve(this.consume(n));
}
if (this.closed) {
return Promise.reject(
this.closeError ?? new FrameDropError('socket_closed', 'Socket closed before read completed'),
);
}
return new Promise<Buffer>((resolve, reject) => {
this.pending = { needed: n, resolve, reject };
});
}
private onData(chunk: Buffer): void {
this.chunks.push(chunk);
this.buffered += chunk.length;
this.tryFlush();
}
private tryFlush(): void {
if (this.pending !== null && this.buffered >= this.pending.needed) {
const { needed, resolve } = this.pending;
this.pending = null;
resolve(this.consume(needed));
}
}
private consume(n: number): Buffer {
// Concatenate chunks into a single contiguous buffer, take n bytes, put any
// remainder back as the first chunk.
const combined = Buffer.concat(this.chunks);
this.chunks.length = 0;
const taken = combined.subarray(0, n);
const rest = combined.subarray(n);
if (rest.length > 0) {
this.chunks.push(rest);
}
this.buffered = rest.length;
return Buffer.from(taken); // copy so callers own the memory
}
private onClose(): void {
this.closed = true;
if (this.pending !== null) {
const { reject } = this.pending;
this.pending = null;
reject(new FrameDropError('socket_closed', 'Socket closed before read completed'));
}
}
private onSocketError(err: Error): void {
this.closed = true;
this.closeError = err;
if (this.pending !== null) {
const { reject } = this.pending;
this.pending = null;
reject(err);
}
}
}
// ---------------------------------------------------------------------------
// Frame envelope reader
// ---------------------------------------------------------------------------
/**
* Reads the next complete AVL frame envelope from the socket via `reader`.
* Validates preamble, length sanity, and CRC. Does NOT dispatch on codec ID —
* that is the session loop's responsibility.
*
* Throws `FrameDropError` when the connection must be closed (invalid preamble,
* oversized frame, N1≠N2). On CRC mismatch the frame is returned with
* `crcValid: false` — the caller should not ACK but should keep the connection.
*/
export async function readNextFrame(reader: BufferedReader): Promise<Frame> {
// 1. Preamble: 4 zero bytes
const preamble = await reader.readExact(4);
if (preamble.readUInt32BE(0) !== 0) {
throw new FrameDropError(
'invalid_preamble',
`Expected preamble 0x00000000, got 0x${preamble.toString('hex')}`,
);
}
// 2. DataFieldLength: number of bytes from CodecID through N2
const lengthBuf = await reader.readExact(4);
const dataFieldLength = lengthBuf.readUInt32BE(0);
if (dataFieldLength < 2 || dataFieldLength > MAX_AVL_PACKET_SIZE) {
throw new FrameDropError(
'oversized_frame',
`DataFieldLength ${dataFieldLength} out of bounds [2, ${MAX_AVL_PACKET_SIZE}]`,
);
}
// 3. Body: CodecID (1B) + N1 (1B) + AVL records + N2 (1B)
const body = await reader.readExact(dataFieldLength);
// 4. CRC field: 4 bytes, value in lower 2 bytes (upper 2 are always 0)
const crcField = await reader.readExact(4);
const expectedCrc = crcField.readUInt16BE(2); // bytes 2–3 carry the 16-bit value
const computedCrc = crc16Ibm(body);
// 5. Validate N1 === N2 (record count repeated for integrity)
// Body layout: [CodecID 1B][N1 1B][...records...][N2 1B]
// N2 is the last byte of body.
if (body.length < 2) {
throw new FrameDropError('oversized_frame', 'Body too short to contain CodecID and N1');
}
const n1 = body[1] ?? 0;
const n2 = body[body.length - 1] ?? 0;
if (n1 !== n2) {
throw new FrameDropError(
'n1_n2_mismatch',
`N1 (${n1}) !== N2 (${n2}): structural mismatch, dropping connection`,
);
}
const codecId = body[0] ?? 0;
return {
codecId,
payload: body,
crcValid: expectedCrc === computedCrc,
expectedCrc,
computedCrc,
};
}
src/adapters/teltonika/handshake.ts
+62
View File
@@ -0,0 +1,62 @@
import type * as net from 'node:net';
import { BufferedReader } from './frame.js';
/**
* IMEI validation pattern. Teltonika devices send 15-digit IMEIs; we allow
* 14–16 to cover edge cases (14-digit test units, 16-digit future variants).
*/
const IMEI_PATTERN = /^\d{14,16}$/;
/**
* Maximum IMEI byte length we accept. Teltonika spec says 15; we allow up to
* 32 as headroom before rejecting as malformed.
*/
const MAX_IMEI_LENGTH = 32;
export class HandshakeError extends Error {
constructor(
message: string,
public readonly rawBytes?: string,
) {
super(message);
this.name = 'HandshakeError';
}
}
/**
* Reads and validates the Teltonika IMEI handshake from the socket.
* Wire format: [length 2B big-endian][IMEI ASCII bytes (length B)]
*
* Returns the IMEI string on success. Does NOT write the accept/reject byte —
* that decision belongs to the session loop after consulting DeviceAuthority.
*
* Throws HandshakeError for any malformed input without writing to the socket.
*/
export async function readImeiHandshake(socket: net.Socket): Promise<string> {
const reader = new BufferedReader(socket);
// 1. Read 2-byte length field
const lengthBuf = await reader.readExact(2);
const imeiLength = lengthBuf.readUInt16BE(0);
if (imeiLength === 0 || imeiLength > MAX_IMEI_LENGTH) {
throw new HandshakeError(
`IMEI length ${imeiLength} is outside valid range [1, ${MAX_IMEI_LENGTH}]`,
lengthBuf.toString('hex'),
);
}
// 2. Read IMEI bytes
const imeiBuf = await reader.readExact(imeiLength);
const imei = imeiBuf.toString('ascii');
// 3. Validate: must be all digits, 14–16 chars
if (!IMEI_PATTERN.test(imei)) {
throw new HandshakeError(
`IMEI "${imei}" does not match expected pattern (14–16 digits)`,
imeiBuf.toString('hex'),
);
}
return imei;
}
src/adapters/teltonika/index.ts
+175
View File
@@ -0,0 +1,175 @@
import type * as net from 'node:net';
import type { Adapter, AdapterContext } from '../../core/types.js';
import type { DeviceAuthority } from './device-authority.js';
import { AllowAllAuthority } from './device-authority.js';
import { readImeiHandshake, HandshakeError } from './handshake.js';
import { BufferedReader, readNextFrame, FrameDropError } from './frame.js';
import { CodecRegistry } from './codec/registry.js';
export type TeltonikaAdapterOptions = {
readonly port: number;
readonly deviceAuthority?: DeviceAuthority;
readonly strictDeviceAuth?: boolean;
readonly codecRegistry?: CodecRegistry;
};
/**
* Creates and returns the Teltonika adapter. The adapter:
* 1. Performs the IMEI handshake (reads; consults DeviceAuthority; writes 0x01/0x00)
* 2. Runs the AVL frame read loop (preamble → length → body → CRC → dispatch)
* 3. ACKs accepted frames with the 4-byte big-endian record count
*
* Codec handlers are registered externally and passed in via `codecRegistry`.
* Tasks 1.5–1.7 populate the registry; this task ships it empty (any frame
* triggers the unknown-codec path and drops the connection, per spec).
*/
export function createTeltonikaAdapter(options: TeltonikaAdapterOptions): Adapter {
const authority: DeviceAuthority = options.deviceAuthority ?? new AllowAllAuthority();
const strictDeviceAuth = options.strictDeviceAuth ?? false;
const codecRegistry = options.codecRegistry ?? new CodecRegistry();
return {
name: 'teltonika',
ports: [options.port],
async handleSession(socket: net.Socket, ctx: AdapterContext): Promise<void> {
// ------------------------------------------------------------------ //
// Phase 1: IMEI handshake
// ------------------------------------------------------------------ //
let imei: string;
try {
imei = await readImeiHandshake(socket);
} catch (err) {
if (err instanceof HandshakeError) {
ctx.logger.warn(
{ err, raw_bytes: err.rawBytes },
'IMEI handshake failed; destroying socket',
);
} else {
ctx.logger.warn({ err }, 'unexpected error during IMEI handshake');
}
socket.destroy();
return;
}
const sessionLogger = ctx.logger.child({ imei });
// Consult DeviceAuthority — errors default to 'unknown' (safe, observable)
let knownLabel: 'known' | 'unknown';
try {
knownLabel = await authority.check(imei);
} catch (authorityErr) {
sessionLogger.warn(
{ err: authorityErr },
'DeviceAuthority.check failed; defaulting to unknown',
);
knownLabel = 'unknown';
}
ctx.metrics.inc('teltonika_handshake_total', {
result: 'accepted',
known: knownLabel,
});
if (knownLabel === 'unknown' && strictDeviceAuth) {
// Reject path (opt-in via STRICT_DEVICE_AUTH)
socket.write(Buffer.from([0x00]));
sessionLogger.warn({ imei }, 'rejected unknown device under STRICT_DEVICE_AUTH');
socket.destroy();
return;
}
// Accept the device
socket.write(Buffer.from([0x01]));
sessionLogger.debug({ known: knownLabel }, 'IMEI handshake accepted');
// ------------------------------------------------------------------ //
// Phase 2: AVL frame read loop
// ------------------------------------------------------------------ //
const reader = new BufferedReader(socket);
while (!socket.destroyed) {
let frame;
try {
frame = await readNextFrame(reader);
} catch (err) {
if (err instanceof FrameDropError) {
if (err.reason === 'socket_closed') {
// Normal disconnect — no warning needed
sessionLogger.debug('socket closed during frame read');
} else {
sessionLogger.warn(
{ reason: err.reason, err },
'malformed frame; dropping connection',
);
}
} else {
sessionLogger.warn({ err }, 'unexpected error reading frame; dropping connection');
}
socket.destroy();
return;
}
if (!frame.crcValid) {
sessionLogger.warn(
{
expected_crc: `0x${frame.expectedCrc.toString(16).padStart(4, '0')}`,
computed_crc: `0x${frame.computedCrc.toString(16).padStart(4, '0')}`,
frame_length: frame.payload.length,
},
'CRC mismatch; not ACKing (device will retransmit)',
);
ctx.metrics.inc('teltonika_frames_total', {
codec: `0x${frame.codecId.toString(16)}`,
result: 'crc_fail',
});
// Do NOT ACK — connection stays open for device retransmit
continue;
}
const handler = codecRegistry.get(frame.codecId);
if (handler === undefined) {
sessionLogger.warn(
{
codec_id: `0x${frame.codecId.toString(16).padStart(2, '0')}`,
header: frame.payload.subarray(0, 16).toString('hex'),
},
'unknown codec; dropping connection',
);
ctx.metrics.inc('teltonika_unknown_codec_total', {
codec_id: `0x${frame.codecId.toString(16).padStart(2, '0')}`,
});
socket.destroy();
return;
}
let result: { recordCount: number };
try {
result = await handler.handle(frame.payload, {
imei,
publish: ctx.publish,
logger: sessionLogger,
});
} catch (handlerErr) {
sessionLogger.warn(
{ err: handlerErr, codec_id: `0x${frame.codecId.toString(16).padStart(2, '0')}` },
'codec handler threw; dropping connection',
);
socket.destroy();
return;
}
ctx.metrics.inc('teltonika_frames_total', {
codec: `0x${frame.codecId.toString(16).padStart(2, '0')}`,
result: 'ok',
});
// ACK: 4-byte big-endian record count
const ack = Buffer.alloc(4);
ack.writeUInt32BE(result.recordCount, 0);
socket.write(ack);
}
},
};
}
src/config/load.ts
+48
View File
@@ -0,0 +1,48 @@
import { randomUUID } from 'node:crypto';
import { z } from 'zod';
const ConfigSchema = z.object({
NODE_ENV: z.enum(['development', 'test', 'production']).default('development'),
INSTANCE_ID: z.string().min(1).default(() => `local-${randomUUID().slice(0, 8)}`),
LOG_LEVEL: z
.enum(['fatal', 'error', 'warn', 'info', 'debug', 'trace'])
.default('info'),
// Vendor port bindings — extend as adapters are added
TELTONIKA_PORT: z.coerce.number().int().min(1).max(65535).default(5027),
// Redis — required in all environments; no silent default
REDIS_URL: z.string().url(),
REDIS_TELEMETRY_STREAM: z.string().min(1).default('telemetry:teltonika'),
REDIS_STREAM_MAXLEN: z.coerce.number().int().min(0).default(1_000_000),
// Observability
METRICS_PORT: z.coerce.number().int().min(0).max(65535).default(9090),
// Device authority — off by default; opt-in for strict reject-on-unknown
STRICT_DEVICE_AUTH: z
.string()
.transform((v) => v === 'true' || v === '1')
.default('false'),
});
export type Config = z.infer<typeof ConfigSchema>;
/**
* Validates process.env at startup and returns a typed Config.
* Throws with a human-readable error listing every missing/invalid key if
* validation fails — the intent is loud, fast failure rather than running
* with bad configuration.
*/
export function loadConfig(env: Record<string, string | undefined> = process.env): Config {
const result = ConfigSchema.safeParse(env);
if (!result.success) {
const issues = result.error.issues
.map((issue) => ` ${issue.path.join('.')}: ${issue.message}`)
.join('\n');
throw new Error(`Configuration error — invalid or missing environment variables:\n${issues}`);
}
return result.data;
}
src/core/publish.ts
+22
View File
@@ -0,0 +1,22 @@
import type { Logger } from 'pino';
import type { Position } from './types.js';
/**
* Stub publish function — logs the position at debug level.
* Real implementation (Redis Streams XADD) lands in task 1.8.
* The signature is already the final shape so adapter types stabilize now.
*/
export function makePublisher(logger: Logger): (position: Position) => Promise<void> {
return async (position: Position): Promise<void> => {
logger.debug(
{
device_id: position.device_id,
timestamp: position.timestamp.toISOString(),
latitude: position.latitude,
longitude: position.longitude,
speed: position.speed,
},
'publish position (stub)',
);
};
}
src/core/registry.ts
+29
View File
@@ -0,0 +1,29 @@
import type { Adapter } from './types.js';
/**
* Maps TCP port numbers to their owning adapter. Populated at startup by main.ts
* before the server begins accepting connections.
*/
export class AdapterRegistry {
private readonly portMap = new Map<number, Adapter>();
register(adapter: Adapter): void {
for (const port of adapter.ports) {
if (this.portMap.has(port)) {
throw new Error(
`Port ${port} is already registered by adapter "${this.portMap.get(port)!.name}". ` +
`Attempted to register adapter "${adapter.name}".`,
);
}
this.portMap.set(port, adapter);
}
}
get(port: number): Adapter | undefined {
return this.portMap.get(port);
}
ports(): ReadonlyArray<number> {
return Array.from(this.portMap.keys());
}
}
src/core/server.ts
+31
View File
@@ -0,0 +1,31 @@
import * as net from 'node:net';
import type { Adapter, AdapterContext } from './types.js';
import { runSession } from './session.js';
/**
* Starts a TCP server for a single adapter. Returns the underlying net.Server
* so callers can attach error listeners and call server.close() on shutdown.
*/
export function startServer(port: number, adapter: Adapter, ctx: AdapterContext): net.Server {
const server = net.createServer((socket) => {
// runSession is async but we must not await it here — that would serialize
// connections. Fire-and-forget with an explicit rejection guard.
runSession(socket, adapter, ctx).catch((err) => {
ctx.logger.warn({ err, adapter: adapter.name }, 'unhandled error escaping runSession');
});
});
server.on('error', (err) => {
ctx.logger.error({ err, port, adapter: adapter.name }, 'TCP server error');
});
server.on('close', () => {
ctx.logger.info({ port, adapter: adapter.name }, 'TCP server closed');
});
server.listen(port, () => {
ctx.logger.info({ port, adapter: adapter.name }, 'TCP server listening');
});
return server;
}
src/core/session.ts
+48
View File
@@ -0,0 +1,48 @@
import type * as net from 'node:net';
import type { Adapter, AdapterContext } from './types.js';
/**
* Wraps a single adapter session with standard socket configuration, error
* handling, and lifecycle logging. Deliberately knows nothing about IMEI,
* framing, or codecs — those are entirely the adapter's business.
*/
export async function runSession(
socket: net.Socket,
adapter: Adapter,
ctx: AdapterContext,
): Promise<void> {
// Disable Nagle algorithm — we send small ACK bytes where latency matters
// more than throughput efficiency.
socket.setNoDelay(true);
// TCP keepalive: detect dead connections (idle NAT timeouts, crashed devices)
// after 60s. Safe because devices already retransmit on reconnect.
socket.setKeepAlive(true, 60_000);
const remoteAddress = `${socket.remoteAddress ?? 'unknown'}:${socket.remotePort ?? '?'}`;
const sessionLogger = ctx.logger.child({ remote_address: remoteAddress });
sessionLogger.debug({ adapter: adapter.name }, 'session opened');
socket.on('error', (err) => {
sessionLogger.debug({ err }, 'socket error');
});
socket.on('end', () => {
sessionLogger.debug('socket end');
});
socket.on('close', (hadError) => {
sessionLogger.debug({ had_error: hadError }, 'socket closed');
});
try {
await adapter.handleSession(socket, { ...ctx, logger: sessionLogger });
} catch (err) {
sessionLogger.warn({ err }, 'session handler threw; destroying socket');
} finally {
if (!socket.destroyed) {
socket.destroy();
}
}
}
src/core/types.ts
+50
View File
@@ -0,0 +1,50 @@
import type { Logger } from 'pino';
import type * as net from 'node:net';
/**
* Normalized GPS position record — the boundary contract between vendor adapters
* and all downstream consumers. Shape is Teltonika-canonical; other adapters must
* map to the same type.
*/
export type Position = {
readonly device_id: string;
readonly timestamp: Date;
readonly latitude: number;
readonly longitude: number;
readonly altitude: number;
readonly angle: number; // heading 0–360°
readonly speed: number; // km/h; 0 may mean "GPS invalid" — preserve verbatim
readonly satellites: number;
readonly priority: 0 | 1 | 2; // 0=Low, 1=High, 2=Panic
readonly attributes: Readonly<Record<string, number | bigint | Buffer>>;
};
/**
* Minimal metrics surface exposed to adapters. Concrete implementation lands in
* task 1.10; this placeholder keeps types stable through tasks 1.2–1.9.
*/
export type Metrics = {
readonly inc: (name: string, labels?: Record<string, string>) => void;
readonly observe: (name: string, value: number, labels?: Record<string, string>) => void;
};
/**
* Narrow context object passed into each adapter's session handler.
* Adapters receive everything they need here; shell internals are not exposed.
*/
export type AdapterContext = {
readonly publish: (position: Position) => Promise<void>;
readonly logger: Logger;
readonly metrics: Metrics;
};
/**
* Vendor adapter interface. Each adapter declares the port(s) it owns and
* implements the session lifecycle. The core shell calls handleSession once per
* accepted TCP connection and never touches vendor-specific framing.
*/
export type Adapter = {
readonly name: string;
readonly ports: readonly number[];
handleSession(socket: net.Socket, ctx: AdapterContext): Promise<void>;
};
src/main.ts
+73
View File
@@ -0,0 +1,73 @@
import { loadConfig } from './config/load.js';
import { createLogger } from './observability/logger.js';
import { makePublisher } from './core/publish.js';
import { startServer } from './core/server.js';
import { createTeltonikaAdapter } from './adapters/teltonika/index.js';
import { AllowAllAuthority } from './adapters/teltonika/device-authority.js';
import { CodecRegistry } from './adapters/teltonika/codec/registry.js';
import type { Metrics } from './core/types.js';
// -------------------------------------------------------------------------
// Startup: validate config (fail fast on bad env), build logger, boot server
// -------------------------------------------------------------------------
let config;
try {
config = loadConfig();
} catch (err) {
// Config validation failures print a human-readable message and exit 1.
// Logger is not available yet — process.stderr is the only output channel.
process.stderr.write(`${err instanceof Error ? err.message : String(err)}\n`);
process.exit(1);
}
const logger = createLogger({
level: config.LOG_LEVEL,
nodeEnv: config.NODE_ENV,
instanceId: config.INSTANCE_ID,
});
logger.info('tcp-ingestion starting');
// Placeholder metrics implementation — replaced in task 1.10
const metrics: Metrics = {
inc: (name, labels) => logger.debug({ metric: name, labels }, 'metric inc'),
observe: (name, value, labels) => logger.debug({ metric: name, value, labels }, 'metric observe'),
};
const publisher = makePublisher(logger);
// Codec registry — empty until tasks 1.5–1.7 register handlers
const codecRegistry = new CodecRegistry();
const teltonikaAdapter = createTeltonikaAdapter({
port: config.TELTONIKA_PORT,
deviceAuthority: new AllowAllAuthority(),
strictDeviceAuth: config.STRICT_DEVICE_AUTH,
codecRegistry,
});
const ctx = {
publish: publisher,
logger,
metrics,
};
const server = startServer(config.TELTONIKA_PORT, teltonikaAdapter, ctx);
// Graceful shutdown
function shutdown(signal: string): void {
logger.info({ signal }, 'shutdown signal received; closing server');
server.close(() => {
logger.info('server closed; exiting');
process.exit(0);
});
// Force exit after 10s if connections are still open
setTimeout(() => {
logger.warn('forced exit after timeout');
process.exit(1);
}, 10_000).unref();
}
process.on('SIGTERM', () => shutdown('SIGTERM'));
process.on('SIGINT', () => shutdown('SIGINT'));
src/observability/logger.ts
+42
View File
@@ -0,0 +1,42 @@
import pino from 'pino';
import type { Logger } from 'pino';
export type { Logger };
/**
* Builds the root pino logger. Called once at startup with the config values.
*
* In development, pino-pretty is used for human-readable output (lazy import
* so it is never bundled in production paths). In test/production, raw JSON is
* emitted — fast and parseable by log aggregators.
*/
export function createLogger(options: {
level: string;
nodeEnv: string;
instanceId: string;
}): Logger {
const { level, nodeEnv, instanceId } = options;
const base = {
service: 'tcp-ingestion',
instance_id: instanceId,
};
if (nodeEnv === 'development') {
return pino({
level,
base,
transport: {
target: 'pino-pretty',
options: {
colorize: true,
translateTime: 'SYS:standard',
ignore: 'pid,hostname',
},
},
});
}
// Production and test: plain JSON — fast, no extra deps
return pino({ level, base });
}
test/config.test.ts
+54
View File
@@ -0,0 +1,54 @@
import { describe, it, expect } from 'vitest';
import { loadConfig } from '../src/config/load.js';
describe('loadConfig', () => {
it('throws when REDIS_URL is unset and names REDIS_URL in the error', () => {
expect(() => loadConfig({})).toThrowError(/REDIS_URL/);
});
it('returns a valid Config with sensible defaults when only REDIS_URL is set in development', () => {
const config = loadConfig({
NODE_ENV: 'development',
REDIS_URL: 'redis://localhost:6379',
});
expect(config.NODE_ENV).toBe('development');
expect(config.REDIS_URL).toBe('redis://localhost:6379');
expect(config.TELTONIKA_PORT).toBe(5027);
expect(config.REDIS_TELEMETRY_STREAM).toBe('telemetry:teltonika');
expect(config.REDIS_STREAM_MAXLEN).toBe(1_000_000);
expect(config.METRICS_PORT).toBe(9090);
expect(config.LOG_LEVEL).toBe('info');
expect(config.STRICT_DEVICE_AUTH).toBe(false);
expect(config.INSTANCE_ID).toMatch(/^local-[0-9a-f]{8}$/);
});
it('parses TELTONIKA_PORT as a number from a string env var', () => {
const config = loadConfig({
REDIS_URL: 'redis://localhost:6379',
TELTONIKA_PORT: '5555',
});
expect(config.TELTONIKA_PORT).toBe(5555);
});
it('enables STRICT_DEVICE_AUTH when set to "true"', () => {
const config = loadConfig({
REDIS_URL: 'redis://localhost:6379',
STRICT_DEVICE_AUTH: 'true',
});
expect(config.STRICT_DEVICE_AUTH).toBe(true);
});
it('rejects an invalid NODE_ENV value', () => {
expect(() =>
loadConfig({
REDIS_URL: 'redis://localhost:6379',
NODE_ENV: 'staging',
}),
).toThrow();
});
it('rejects a non-URL REDIS_URL', () => {
expect(() => loadConfig({ REDIS_URL: 'not-a-url' })).toThrowError(/REDIS_URL/);
});
});
test/core/server.test.ts
+74
View File
@@ -0,0 +1,74 @@
import { describe, it, expect, vi, afterEach } from 'vitest';
import * as net from 'node:net';
import type { Logger } from 'pino';
import type { Adapter, AdapterContext, Metrics, Position } from '../../src/core/types.js';
import { startServer } from '../../src/core/server.js';
function makeMockContext(): AdapterContext {
const logger = {
debug: vi.fn(),
info: vi.fn(),
warn: vi.fn(),
error: vi.fn(),
child: vi.fn().mockReturnThis(),
} as unknown as Logger;
const metrics: Metrics = {
inc: vi.fn(),
observe: vi.fn(),
};
return {
publish: vi.fn(async (_p: Position) => {}),
logger,
metrics,
};
}
describe('startServer', () => {
const servers: net.Server[] = [];
afterEach(() => {
for (const server of servers) {
server.close();
}
servers.length = 0;
});
it('invokes handleSession with a real socket when a client connects', async () => {
const handleSession = vi.fn().mockResolvedValue(undefined);
const adapter: Adapter = {
name: 'test-adapter',
ports: [0], // port 0 = OS-assigned ephemeral port
handleSession,
};
const ctx = makeMockContext();
const server = startServer(0, adapter, ctx);
servers.push(server);
// Wait for the server to start listening
const port = await new Promise<number>((resolve) => {
server.on('listening', () => {
const addr = server.address();
resolve(typeof addr === 'object' && addr !== null ? addr.port : 0);
});
});
// Connect a client
const client = new net.Socket();
client.connect(port, '127.0.0.1');
// Wait for handleSession to be called
await vi.waitFor(() => expect(handleSession).toHaveBeenCalledOnce(), { timeout: 2000 });
const [socketArg, ctxArg] = handleSession.mock.calls[0] as [net.Socket, AdapterContext];
expect(socketArg).toBeInstanceOf(net.Socket);
expect(ctxArg).toBeDefined();
expect(typeof ctxArg.publish).toBe('function');
expect(ctxArg.logger).toBeDefined();
client.destroy();
});
});
test/crc.test.ts
+85
View File
@@ -0,0 +1,85 @@
import { describe, it, expect } from 'vitest';
import { crc16Ibm } from '../src/adapters/teltonika/crc.js';
describe('crc16Ibm', () => {
it('returns 0x0000 for an empty buffer', () => {
expect(crc16Ibm(Buffer.alloc(0))).toBe(0x0000);
});
it('matches the canonical Teltonika Codec 8 first example (0xC7CF)', () => {
// Full frame (hex stream from the Teltonika doc, first example):
// 000000000000003608010000016B40D8EA30010000000000000000000000000000000105021503010101425E0F01F10000601A014E0000000000000000010000C7CF
//
// Structure:
// 00 00 00 00 preamble
// 00 00 00 36 DataFieldLength = 0x36 = 54
// 08 01 ... body (CodecID through N2) — 54 bytes
// 00 00 C7 CF CRC field
//
// CRC is computed over the 54-byte body only.
const body = Buffer.from(
'08' + // CodecID
'01' + // N1 = 1
'0000016B40D8EA30' + // Timestamp
'01' + // Priority
'00000000' + // Longitude
'00000000' + // Latitude
'0000' + // Altitude
'0000' + // Angle
'00' + // Satellites
'0000' + // Speed
'01' + // Event IO ID
'05' + // N total IO
'02' + // N1 IO (1-byte values)
'15' + '03' + // IO 21 = 3
'01' + '01' + // IO 1 = 1
'01' + // N2 IO (2-byte values)
'42' + '5E0F' + // IO 66 = 0x5E0F
'01' + // N4 IO (4-byte values)
'F1' + '0000601A' + // IO 241 = 0x0000601A
'01' + // N8 IO (8-byte values)
'4E' + '0000000000000000' + // IO 78 = 0
'01', // N2 = 1
'hex',
);
expect(body.length).toBe(0x36); // sanity-check: must be 54 bytes
expect(crc16Ibm(body)).toBe(0xc7cf);
});
it('matches the second Teltonika Codec 8 example (0xF22A)', () => {
// Full frame hex:
// 000000000000002808010000016B40D9AD80010000000000000000000000000000000103021503010101425E100000010000F22A
// Body = bytes from CodecID through N2 (0x28 = 40 bytes)
const body = Buffer.from(
'08' +
'01' +
'0000016B40D9AD80' +
'01' +
'00000000' +
'00000000' +
'0000' +
'0000' +
'00' +
'0000' +
'01' +
'03' +
'02' +
'15' + '03' +
'01' + '01' +
'01' +
'42' + '5E10' +
'00' +
'00' +
'01',
'hex',
);
expect(body.length).toBe(0x28); // 40 bytes
expect(crc16Ibm(body)).toBe(0xf22a);
});
it('produces a non-zero CRC for a single 0xFF byte', () => {
expect(crc16Ibm(Buffer.from([0xff]))).not.toBe(0);
});
});
test/device-authority.test.ts
+19
View File
@@ -0,0 +1,19 @@
import { describe, it, expect } from 'vitest';
import { AllowAllAuthority } from '../src/adapters/teltonika/device-authority.js';
describe('AllowAllAuthority', () => {
it('returns "known" for any IMEI', async () => {
const authority = new AllowAllAuthority();
expect(await authority.check('356307042441013')).toBe('known');
});
it('returns "known" for an empty string (authority ignores content)', async () => {
const authority = new AllowAllAuthority();
expect(await authority.check('')).toBe('known');
});
it('returns "known" for an unknown/unseen IMEI', async () => {
const authority = new AllowAllAuthority();
expect(await authority.check('999999999999999')).toBe('known');
});
});
test/fixtures/teltonika/codec16/.gitkeep
Vendored
View File
test/fixtures/teltonika/codec8/.gitkeep
Vendored
View File
test/fixtures/teltonika/codec8e/.gitkeep
Vendored
View File
test/frame.test.ts
+218
View File
@@ -0,0 +1,218 @@
import { describe, it, expect, beforeEach, vi } from 'vitest';
import { EventEmitter } from 'node:events';
import type * as net from 'node:net';
import { BufferedReader, readNextFrame, FrameDropError } from '../src/adapters/teltonika/frame.js';
import { crc16Ibm } from '../src/adapters/teltonika/crc.js';
// ---------------------------------------------------------------------------
// Minimal mock socket
// ---------------------------------------------------------------------------
function makeMockSocket(): net.Socket & { push(buf: Buffer): void; simulateClose(): void } {
const emitter = new EventEmitter() as net.Socket & {
push(buf: Buffer): void;
simulateClose(): void;
destroyed: boolean;
remoteAddress?: string;
remotePort?: number;
setNoDelay: ReturnType<typeof vi.fn>;
setKeepAlive: ReturnType<typeof vi.fn>;
write: ReturnType<typeof vi.fn>;
destroy: ReturnType<typeof vi.fn>;
};
emitter.destroyed = false;
emitter.remoteAddress = '127.0.0.1';
emitter.remotePort = 12345;
emitter.setNoDelay = vi.fn();
emitter.setKeepAlive = vi.fn();
emitter.write = vi.fn().mockReturnValue(true);
emitter.destroy = vi.fn(() => {
emitter.destroyed = true;
emitter.emit('close', false);
});
emitter.push = (buf: Buffer) => emitter.emit('data', buf);
emitter.simulateClose = () => {
emitter.destroyed = true;
emitter.emit('close', false);
};
return emitter;
}
// ---------------------------------------------------------------------------
// BufferedReader tests
// ---------------------------------------------------------------------------
describe('BufferedReader.readExact', () => {
it('resolves immediately when enough bytes are already buffered', async () => {
const socket = makeMockSocket();
const reader = new BufferedReader(socket);
socket.push(Buffer.from([1, 2, 3, 4, 5]));
const result = await reader.readExact(3);
expect(result).toEqual(Buffer.from([1, 2, 3]));
});
it('correctly accumulates bytes across multiple data events', async () => {
const socket = makeMockSocket();
const reader = new BufferedReader(socket);
const readPromise = reader.readExact(6);
// Send data in three separate chunks
socket.push(Buffer.from([0xaa, 0xbb]));
socket.push(Buffer.from([0xcc, 0xdd]));
socket.push(Buffer.from([0xee, 0xff]));
const result = await readPromise;
expect(result).toEqual(Buffer.from([0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff]));
});
it('handles byte-by-byte arrival', async () => {
const socket = makeMockSocket();
const reader = new BufferedReader(socket);
const readPromise = reader.readExact(4);
for (const byte of [0x01, 0x02, 0x03, 0x04]) {
socket.push(Buffer.from([byte]));
}
const result = await readPromise;
expect(result).toEqual(Buffer.from([0x01, 0x02, 0x03, 0x04]));
});
it('leaves unconsumed bytes available for subsequent reads', async () => {
const socket = makeMockSocket();
const reader = new BufferedReader(socket);
socket.push(Buffer.from([1, 2, 3, 4, 5, 6]));
const first = await reader.readExact(3);
const second = await reader.readExact(3);
expect(first).toEqual(Buffer.from([1, 2, 3]));
expect(second).toEqual(Buffer.from([4, 5, 6]));
});
it('rejects with FrameDropError when socket closes before read completes', async () => {
const socket = makeMockSocket();
const reader = new BufferedReader(socket);
const readPromise = reader.readExact(10);
socket.push(Buffer.from([1, 2, 3])); // only 3 of 10 bytes
socket.simulateClose();
await expect(readPromise).rejects.toBeInstanceOf(FrameDropError);
});
});
// ---------------------------------------------------------------------------
// Frame builder helper (ESM-safe, no require())
// ---------------------------------------------------------------------------
function buildFrame(options: {
codecId?: number;
n1?: number;
n2?: number;
records?: Buffer;
crcOverride?: number;
}): Buffer {
const codecId = options.codecId ?? 0x08;
const n1 = options.n1 ?? 1;
const n2 = options.n2 ?? n1; // default N2 = N1 for valid frames
const records = options.records ?? Buffer.alloc(0);
// Body = CodecID (1B) + N1 (1B) + records + N2 (1B)
const body = Buffer.concat([Buffer.from([codecId, n1]), records, Buffer.from([n2])]);
const realCrc = crc16Ibm(body);
const crc = options.crcOverride !== undefined ? options.crcOverride : realCrc;
const preamble = Buffer.alloc(4, 0);
const lengthBuf = Buffer.alloc(4);
lengthBuf.writeUInt32BE(body.length, 0);
const crcBuf = Buffer.alloc(4);
crcBuf.writeUInt16BE(0, 0); // upper 2 bytes always 0
crcBuf.writeUInt16BE(crc, 2);
return Buffer.concat([preamble, lengthBuf, body, crcBuf]);
}
async function feedFrameToReader(frameBytes: Buffer): Promise<ReturnType<typeof readNextFrame>> {
const socket = makeMockSocket();
const reader = new BufferedReader(socket);
const readPromise = readNextFrame(reader);
socket.push(frameBytes);
return readPromise;
}
// ---------------------------------------------------------------------------
// readNextFrame tests
// ---------------------------------------------------------------------------
describe('readNextFrame', () => {
beforeEach(() => {
vi.clearAllMocks();
});
it('parses a minimal valid frame with correct CRC', async () => {
const frameBytes = buildFrame({ codecId: 0x08, n1: 1, records: Buffer.alloc(0) });
const frame = await feedFrameToReader(frameBytes);
expect(frame.codecId).toBe(0x08);
expect(frame.crcValid).toBe(true);
});
it('reports crcValid=false on CRC mismatch without throwing', async () => {
const frameBytes = buildFrame({ codecId: 0x08, n1: 1, crcOverride: 0xdead });
const frame = await feedFrameToReader(frameBytes);
expect(frame.crcValid).toBe(false);
expect(frame.expectedCrc).toBe(0xdead);
});
it('throws FrameDropError(invalid_preamble) when preamble is non-zero', async () => {
const socket = makeMockSocket();
const reader = new BufferedReader(socket);
const readPromise = readNextFrame(reader);
const badFrame = Buffer.concat([
Buffer.from([0x00, 0x00, 0x00, 0x01]), // bad preamble
Buffer.from([0x00, 0x00, 0x00, 0x03]), // length
Buffer.from([0x08, 0x01, 0x01]), // body
Buffer.from([0x00, 0x00, 0x00, 0x00]), // CRC
]);
socket.push(badFrame);
await expect(readPromise).rejects.toMatchObject({
reason: 'invalid_preamble',
});
});
it('throws FrameDropError(n1_n2_mismatch) when N1 ≠ N2', async () => {
const frameBytes = buildFrame({ codecId: 0x08, n1: 2, n2: 1 });
await expect(feedFrameToReader(frameBytes)).rejects.toMatchObject({
reason: 'n1_n2_mismatch',
});
});
it('handles frame bytes arriving across multiple data events', async () => {
const frameBytes = buildFrame({ codecId: 0x08, n1: 1 });
const socket = makeMockSocket();
const reader = new BufferedReader(socket);
const readPromise = readNextFrame(reader);
// Split frame into 3-byte chunks
for (let offset = 0; offset < frameBytes.length; offset += 3) {
socket.push(frameBytes.subarray(offset, offset + 3));
}
const frame = await readPromise;
expect(frame.crcValid).toBe(true);
expect(frame.codecId).toBe(0x08);
});
});
test/handshake.test.ts
+122
View File
@@ -0,0 +1,122 @@
import { describe, it, expect, vi } from 'vitest';
import { EventEmitter } from 'node:events';
import type * as net from 'node:net';
import { readImeiHandshake, HandshakeError } from '../src/adapters/teltonika/handshake.js';
function makeMockSocket(): net.Socket & { push(buf: Buffer): void } {
const emitter = new EventEmitter() as net.Socket & {
push(buf: Buffer): void;
destroyed: boolean;
write: ReturnType<typeof vi.fn>;
destroy: ReturnType<typeof vi.fn>;
};
emitter.destroyed = false;
emitter.write = vi.fn().mockReturnValue(true);
emitter.destroy = vi.fn(() => {
emitter.destroyed = true;
emitter.emit('close', false);
});
emitter.push = (buf: Buffer) => emitter.emit('data', buf);
return emitter;
}
/**
* Encodes a Teltonika IMEI handshake frame:
* [2B length big-endian][IMEI ASCII bytes]
*/
function encodeImeiHandshake(imei: string): Buffer {
const imeiBytes = Buffer.from(imei, 'ascii');
const header = Buffer.alloc(2);
header.writeUInt16BE(imeiBytes.length, 0);
return Buffer.concat([header, imeiBytes]);
}
describe('readImeiHandshake', () => {
it('parses a valid 15-digit IMEI without writing to the socket', async () => {
const socket = makeMockSocket();
const imei = '356307042441013';
const handshakeBytes = encodeImeiHandshake(imei);
const resultPromise = readImeiHandshake(socket);
socket.push(handshakeBytes);
const result = await resultPromise;
expect(result).toBe(imei);
// Must NOT write anything — accept/reject is the session loop's job
expect(socket.write).not.toHaveBeenCalled();
});
it('parses a valid IMEI arriving in two chunks (header + body split)', async () => {
const socket = makeMockSocket();
const imei = '123456789012345';
const handshakeBytes = encodeImeiHandshake(imei);
const resultPromise = readImeiHandshake(socket);
// Split: first push just the 2-byte header, then the IMEI bytes
socket.push(handshakeBytes.subarray(0, 2));
socket.push(handshakeBytes.subarray(2));
const result = await resultPromise;
expect(result).toBe(imei);
});
it('parses the Teltonika doc example IMEI (356307042441013)', async () => {
// From the doc: IMEI 356307042441013 encoded as 000F333536333037303432343431303133
const socket = makeMockSocket();
const frame = Buffer.from('000F333536333037303432343431303133', 'hex');
const resultPromise = readImeiHandshake(socket);
socket.push(frame);
const result = await resultPromise;
expect(result).toBe('356307042441013');
});
it('throws HandshakeError for IMEI with non-digit characters', async () => {
const socket = makeMockSocket();
const badImei = 'ABCDEFGHIJKLMNO'; // 15 chars, not digits
const frame = encodeImeiHandshake(badImei);
const resultPromise = readImeiHandshake(socket);
socket.push(frame);
await expect(resultPromise).rejects.toBeInstanceOf(HandshakeError);
expect(socket.write).not.toHaveBeenCalled();
});
it('throws HandshakeError for IMEI that is too short (< 14 digits)', async () => {
const socket = makeMockSocket();
const shortImei = '1234567890123'; // 13 digits
const frame = encodeImeiHandshake(shortImei);
const resultPromise = readImeiHandshake(socket);
socket.push(frame);
await expect(resultPromise).rejects.toBeInstanceOf(HandshakeError);
});
it('throws HandshakeError for zero-length IMEI', async () => {
const socket = makeMockSocket();
const frame = Buffer.from([0x00, 0x00]); // length = 0
const resultPromise = readImeiHandshake(socket);
socket.push(frame);
await expect(resultPromise).rejects.toBeInstanceOf(HandshakeError);
expect(socket.write).not.toHaveBeenCalled();
});
it('throws HandshakeError for excessively long IMEI length field (> 32)', async () => {
const socket = makeMockSocket();
const frame = Buffer.alloc(2);
frame.writeUInt16BE(33, 0); // length = 33
const resultPromise = readImeiHandshake(socket);
socket.push(frame);
await expect(resultPromise).rejects.toBeInstanceOf(HandshakeError);
});
});
test/teltonika-session.test.ts
+282
View File
@@ -0,0 +1,282 @@
import { describe, it, expect, vi, beforeEach } from 'vitest';
import { EventEmitter } from 'node:events';
import type * as net from 'node:net';
import type { Logger } from 'pino';
import type { AdapterContext, Metrics, Position } from '../src/core/types.js';
import { createTeltonikaAdapter } from '../src/adapters/teltonika/index.js';
import { CodecRegistry } from '../src/adapters/teltonika/codec/registry.js';
import type { DeviceAuthority } from '../src/adapters/teltonika/device-authority.js';
import { crc16Ibm } from '../src/adapters/teltonika/crc.js';
// ---------------------------------------------------------------------------
// Mock helpers
// ---------------------------------------------------------------------------
type MockSocket = net.Socket & {
push(buf: Buffer): void;
simulateClose(): void;
getWritten(): Buffer[];
};
function makeMockSocket(): MockSocket {
const written: Buffer[] = [];
const emitter = new EventEmitter() as MockSocket & {
destroyed: boolean;
write: ReturnType<typeof vi.fn>;
destroy: ReturnType<typeof vi.fn>;
setNoDelay: ReturnType<typeof vi.fn>;
setKeepAlive: ReturnType<typeof vi.fn>;
remoteAddress: string;
remotePort: number;
};
emitter.destroyed = false;
emitter.remoteAddress = '127.0.0.1';
emitter.remotePort = 9999;
emitter.setNoDelay = vi.fn();
emitter.setKeepAlive = vi.fn();
emitter.write = vi.fn((data: Buffer | string) => {
written.push(Buffer.isBuffer(data) ? data : Buffer.from(data));
return true;
});
emitter.destroy = vi.fn(() => {
if (!emitter.destroyed) {
emitter.destroyed = true;
emitter.emit('close', false);
}
});
emitter.push = (buf: Buffer) => emitter.emit('data', buf);
emitter.simulateClose = () => {
if (!emitter.destroyed) {
emitter.destroyed = true;
emitter.emit('close', false);
}
};
emitter.getWritten = () => written;
return emitter;
}
function makeMockContext(): AdapterContext {
const logger = {
debug: vi.fn(),
info: vi.fn(),
warn: vi.fn(),
error: vi.fn(),
child: vi.fn().mockReturnThis(),
} as unknown as Logger;
const metrics: Metrics = {
inc: vi.fn(),
observe: vi.fn(),
};
return {
publish: vi.fn(async (_p: Position) => {}),
logger,
metrics,
};
}
/**
* Encodes the Teltonika IMEI handshake wire format.
*/
function encodeImei(imei: string): Buffer {
const body = Buffer.from(imei, 'ascii');
const header = Buffer.alloc(2);
header.writeUInt16BE(body.length, 0);
return Buffer.concat([header, body]);
}
/**
* Builds a complete AVL frame. Body is CodecID + N1 + empty records + N2.
*/
function buildFrame(options: { codecId: number; recordCount?: number; crcOverride?: number }): Buffer {
const n = options.recordCount ?? 1;
const body = Buffer.from([options.codecId, n, n]);
const realCrc = crc16Ibm(body);
const crc = options.crcOverride !== undefined ? options.crcOverride : realCrc;
const preamble = Buffer.alloc(4, 0);
const lengthBuf = Buffer.alloc(4);
lengthBuf.writeUInt32BE(body.length, 0);
const crcBuf = Buffer.alloc(4);
crcBuf.writeUInt16BE(0, 0);
crcBuf.writeUInt16BE(crc, 2);
return Buffer.concat([preamble, lengthBuf, body, crcBuf]);
}
// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------
const IMEI = '356307042441013';
describe('Teltonika adapter — session', () => {
beforeEach(() => {
vi.clearAllMocks();
});
describe('unknown codec path', () => {
it('destroys the socket without writing an ACK when codec ID is not registered', async () => {
const socket = makeMockSocket();
const ctx = makeMockContext();
const registry = new CodecRegistry(); // empty — no codecs registered
const adapter = createTeltonikaAdapter({ port: 5027, codecRegistry: registry });
const sessionPromise = adapter.handleSession(socket, ctx);
// Push IMEI handshake — session writes 0x01 accept
socket.push(encodeImei(IMEI));
// Wait until the 0x01 accept byte is written
await vi.waitFor(
() => {
expect(socket.getWritten().length).toBeGreaterThan(0);
},
{ timeout: 2000 },
);
// Push a frame with an unregistered codec (0x99)
socket.push(buildFrame({ codecId: 0x99 }));
// Session should destroy the socket after unknown codec
await sessionPromise;
expect(socket.destroy).toHaveBeenCalled();
// The ONLY write must be the 0x01 accept byte — no 4-byte ACK
const allWritten = Buffer.concat(socket.getWritten());
expect(allWritten).toEqual(Buffer.from([0x01]));
}, 10_000);
});
describe('CRC mismatch path', () => {
it('does NOT write an ACK on CRC mismatch and keeps the socket open', async () => {
const socket = makeMockSocket();
const ctx = makeMockContext();
const registry = new CodecRegistry();
const adapter = createTeltonikaAdapter({ port: 5027, codecRegistry: registry });
const sessionPromise = adapter.handleSession(socket, ctx);
// Handshake
socket.push(encodeImei(IMEI));
// Wait for the accept byte
await vi.waitFor(
() => {
expect(socket.getWritten().length).toBeGreaterThan(0);
},
{ timeout: 2000 },
);
const writtenBeforeFrame = socket.getWritten().length;
// Push a frame with a deliberately wrong CRC
socket.push(buildFrame({ codecId: 0x08, crcOverride: 0xdead }));
// Give the event loop time to process the frame
await new Promise<void>((resolve) => setTimeout(resolve, 100));
// Socket must still be open
expect(socket.destroy).not.toHaveBeenCalled();
expect(socket.destroyed).toBe(false);
// No new writes after the frame (no ACK sent)
expect(socket.getWritten().length).toBe(writtenBeforeFrame);
// Clean up — simulate device disconnect
socket.simulateClose();
await sessionPromise;
}, 10_000);
});
describe('STRICT_DEVICE_AUTH', () => {
it('writes 0x00 and destroys socket for unknown device when STRICT_DEVICE_AUTH=true', async () => {
const socket = makeMockSocket();
const ctx = makeMockContext();
const strictAuthority: DeviceAuthority = {
check: vi.fn().mockResolvedValue('unknown'),
};
const adapter = createTeltonikaAdapter({
port: 5027,
deviceAuthority: strictAuthority,
strictDeviceAuth: true,
});
const sessionPromise = adapter.handleSession(socket, ctx);
socket.push(encodeImei(IMEI));
await sessionPromise;
const allWritten = Buffer.concat(socket.getWritten());
expect(allWritten).toEqual(Buffer.from([0x00]));
expect(socket.destroy).toHaveBeenCalled();
});
it('accepts unknown device when STRICT_DEVICE_AUTH=false (default)', async () => {
const socket = makeMockSocket();
const ctx = makeMockContext();
const unknownAuthority: DeviceAuthority = {
check: vi.fn().mockResolvedValue('unknown'),
};
const adapter = createTeltonikaAdapter({
port: 5027,
deviceAuthority: unknownAuthority,
strictDeviceAuth: false,
});
const sessionPromise = adapter.handleSession(socket, ctx);
socket.push(encodeImei(IMEI));
// Wait for the accept byte
await vi.waitFor(
() => {
expect(socket.getWritten().length).toBeGreaterThan(0);
},
{ timeout: 2000 },
);
const allWritten = Buffer.concat(socket.getWritten());
expect(allWritten[0]).toBe(0x01);
// Not destroyed
expect(socket.destroy).not.toHaveBeenCalled();
socket.simulateClose();
await sessionPromise;
}, 10_000);
});
describe('known device acceptance', () => {
it('writes 0x01 for a known device and stays connected', async () => {
const socket = makeMockSocket();
const ctx = makeMockContext();
const adapter = createTeltonikaAdapter({ port: 5027 });
const sessionPromise = adapter.handleSession(socket, ctx);
socket.push(encodeImei(IMEI));
// Wait for the accept byte
await vi.waitFor(
() => {
expect(socket.getWritten().length).toBeGreaterThan(0);
},
{ timeout: 2000 },
);
const allWritten = Buffer.concat(socket.getWritten());
expect(allWritten[0]).toBe(0x01);
socket.simulateClose();
await sessionPromise;
}, 10_000);
});
});
tsconfig.json
+21
View File
@@ -0,0 +1,21 @@
{
"compilerOptions": {
"target": "ES2022",
"module": "NodeNext",
"moduleResolution": "NodeNext",
"lib": ["ES2022"],
"outDir": "dist",
"rootDir": "src",
"strict": true,
"noUncheckedIndexedAccess": true,
"noImplicitOverride": true,
"noFallthroughCasesInSwitch": true,
"declaration": false,
"skipLibCheck": true,
"esModuleInterop": false,
"forceConsistentCasingInFileNames": true,
"resolveJsonModule": true
},
"include": ["src/**/*"],
"exclude": ["node_modules", "dist", "test"]
}
tsconfig.test.json
+19
View File
@@ -0,0 +1,19 @@
{
"compilerOptions": {
"target": "ES2022",
"module": "NodeNext",
"moduleResolution": "NodeNext",
"lib": ["ES2022"],
"strict": true,
"noUncheckedIndexedAccess": true,
"noImplicitOverride": true,
"noFallthroughCasesInSwitch": true,
"declaration": false,
"skipLibCheck": true,
"esModuleInterop": false,
"forceConsistentCasingInFileNames": true,
"resolveJsonModule": true,
"noEmit": true
},
"include": ["src/**/*", "test/**/*"]
}
vitest.config.ts
+18
View File
@@ -0,0 +1,18 @@
import { defineConfig } from 'vitest/config';
export default defineConfig({
test: {
include: ['test/**/*.test.ts'],
environment: 'node',
coverage: {
provider: 'v8',
reporter: ['text', 'lcov'],
include: ['src/**/*.ts'],
},
},
resolve: {
// Allow vitest to import .ts files without explicit extensions
// when referenced from test files that don't use .js suffixes
extensions: ['.ts', '.js'],
},
});