1import type * as Effect from "effect/Effect";
2import type * as Stream from "effect/Stream";
3import * as Binding from "../../Binding.ts";
4import type { Destination } from "./Destination.ts";
5
6/**
7 * The JSON envelope AWS IoT Core for LoRaWAN delivers to a `RuleName`
8 * destination's IoT rule for every uplink — the raw payload (base64) plus
9 * the radio metadata.
10 */
11export interface WirelessUplinkMessage {
12 /** ID of the wireless device that sent the uplink. */
13 WirelessDeviceId: string;
14 /** The uplink payload, base64-encoded. */
15 PayloadData: string;
16 /** Radio metadata (DevEui, FPort, data rate, gateway RSSI/SNR, ...). */
17 WirelessMetadata?: {
18 /** LoRaWAN uplink metadata. */
19 LoRaWAN?: {
20 /** The DevEui of the sending device. */
21 DevEui?: string;
22 /** The FPort the payload was sent on. */
23 FPort?: number;
24 /** ISO-8601 server-side receive timestamp. */
25 Timestamp?: string;
26 [key: string]: unknown;
27 };
28 /** Amazon Sidewalk uplink metadata. */
29 Sidewalk?: { [key: string]: unknown };
30 };
31 [key: string]: unknown;
32}
33
34/**
35 * The handler invoked with a stream of uplink messages routed through the
36 * bound destination.
37 */
38export type WirelessUplinkHandlerFn<Req = never> = (
39 stream: Stream.Stream<WirelessUplinkMessage>,
40) => Effect.Effect<void, never, Req>;
41
42export type DestinationEventSourceService = <Req = never>(
43 destination: Destination,
44 process: WirelessUplinkHandlerFn<Req>,
45) => Effect.Effect<void, never, never>;
46
47/**
48 * Event source connecting an IoT Wireless {@link Destination} to the
49 * hosting compute — every uplink a wireless device sends through the
50 * destination invokes the handler.
51 *
52 * The destination must use `expressionType: "RuleName"`. At deploy time the
53 * Lambda implementation (`Lambda.WirelessDestinationEventSource`) creates
54 * the IoT topic rule the destination's `expression` names, with a Lambda
55 * action targeting the current function, and grants `iot.amazonaws.com`
56 * permission to invoke it; at runtime it dispatches uplink invocations to
57 * the handler.
58 *
59 * Use the {@link consumeUplinks} helper rather than the service directly,
60 * and provide `Lambda.WirelessDestinationEventSource` on the hosting
61 * function.
62 * ### Consuming Uplinks
63 * **Example:** Consume LoRaWAN Uplinks in a Lambda
64 * ```typescript
65 * export default IngestFunction.make(
66 * { main: import.meta.url },
67 * Effect.gen(function* () {
68 * const destination = yield* IoTWireless.Destination("Uplinks", {
69 * expressionType: "RuleName",
70 * expression: "sensor_uplinks",
71 * roleArn: deliveryRole.roleArn,
72 * });
73 *
74 * // deploy: creates the `sensor_uplinks` IoT rule targeting this Lambda
75 * // runtime: handles every uplink routed through the destination
76 * yield* IoTWireless.consumeUplinks(destination, (uplinks) =>
77 * uplinks.pipe(
78 * Stream.runForEach((uplink) =>
79 * Effect.log(uplink.WirelessDeviceId, uplink.PayloadData),
80 * ),
81 * Effect.orDie,
82 * ),
83 * );
84 *
85 * return {};
86 * }).pipe(Effect.provide(Lambda.WirelessDestinationEventSource)),
87 * );
88 * ```
89 *
90 * @binding
91 */
92export interface DestinationEventSource extends Binding.Service<
93 DestinationEventSource,
94 "AWS.IoTWireless.DestinationEventSource",
95 DestinationEventSourceService
96> {}
97
98export const DestinationEventSource = Binding.Service<DestinationEventSource>(
99 "AWS.IoTWireless.DestinationEventSource",
100);
101
102/**
103 * Invoke an Effect handler for every uplink message routed through the
104 * destination, by creating the IoT rule the destination's `expression`
105 * names with a Lambda action targeting the current function.
106 *
107 * Provide `Lambda.WirelessDestinationEventSource` on the hosting function
108 * to satisfy the requirement.
109 *
110 * @param destination The `RuleName` destination whose uplinks to consume.
111 * @param process The handler invoked with a stream of uplink messages.
112 *
113 * @example Store uplinks in DynamoDB
114 * ```typescript
115 * yield* IoTWireless.consumeUplinks(destination, (uplinks) =>
116 * uplinks.pipe(
117 * Stream.runForEach((uplink) =>
118 * putItem({ Item: { pk: { S: uplink.WirelessDeviceId } } }),
119 * ),
120 * Effect.orDie,
121 * ),
122 * );
123 * ```
124 */
125export function consumeUplinks<Req = never>(
126 destination: Destination,
127 process: WirelessUplinkHandlerFn<Req>,
128): Effect.Effect<void, never, DestinationEventSource> {
129 return DestinationEventSource.use((source) => source(destination, process));
130}