1import * as Effect from "effect/Effect";
2import * as Layer from "effect/Layer";
3import * as Stream from "effect/Stream";
4import * as Namespace from "../../Namespace.ts";
5import { TopicRule } from "../IoT/TopicRule.ts";
6import type { Destination } from "../IoTWireless/Destination.ts";
7import {
8 DestinationEventSource as IoTWirelessDestinationEventSource,
9 type DestinationEventSourceService,
10 type WirelessUplinkHandlerFn,
11 type WirelessUplinkMessage,
12} from "../IoTWireless/DestinationEventSource.ts";
13import * as Lambda from "./Function.ts";
14import { Permission as LambdaPermission } from "./Permission.ts";
15
16/**
17 * An IoT Wireless uplink invocation — the envelope IoT Core for LoRaWAN
18 * delivers to a destination's rule: the base64 `PayloadData` plus the
19 * sending device's id.
20 */
21export const isWirelessUplinkMessage = (
22 event: any,
23): event is WirelessUplinkMessage =>
24 event != null &&
25 typeof event === "object" &&
26 typeof event.WirelessDeviceId === "string" &&
27 typeof event.PayloadData === "string";
28
29/**
30 * Connects an IoT Wireless {@link Destination}'s uplink traffic to the
31 * current Lambda function.
32 *
33 * At deploy time this layer creates the IoT topic rule named by the
34 * destination's `expression` (the destination must use
35 * `expressionType: "RuleName"`) with a Lambda action targeting this
36 * function, and grants `iot.amazonaws.com` permission to invoke it; at
37 * runtime it dispatches uplink invocations to the registered handler.
38 * ### Consuming wireless uplinks
39 * **Example:** Consume LoRaWAN uplinks
40 * ```typescript
41 * yield* IoTWireless.consumeUplinks(destination, (uplinks) =>
42 * uplinks.pipe(
43 * Stream.runForEach((uplink) => Effect.log(uplink.PayloadData)),
44 * Effect.orDie,
45 * ),
46 * );
47 * ```
48 *
49 * @binding
50 */
51export const WirelessDestinationEventSource = Layer.effect(
52 IoTWirelessDestinationEventSource,
53 // @effect-diagnostics-next-line missingEffectContext:off
54 Effect.gen(function* () {
55 // this layer can only be used in a Lambda Function
56 const host = yield* Lambda.Function;
57 const Rule = yield* TopicRule;
58 const Permission = yield* LambdaPermission;
59
60 return Effect.fn(function* <Req = never>(
61 destination: Destination,
62 process: WirelessUplinkHandlerFn<Req>,
63 ) {
64 // Deploy-time: create the IoT topic rule the destination's expression
65 // names, with a Lambda action targeting this function, and grant
66 // iot.amazonaws.com permission to invoke it (scoped to the rule ARN).
67 // IoT Wireless invokes the rule directly by name, so the SQL FROM
68 // clause is never matched against a real topic. Skipped once running
69 // inside the deployed function, where the only work is registering
70 // the runtime handler.
71 if (!globalThis.__ALCHEMY_RUNTIME__) {
72 yield* Namespace.push(
73 host.LogicalId,
74 Effect.gen(function* () {
75 const rule = yield* Rule(`${destination.LogicalId}-UplinkRule`, {
76 ruleName: destination.expression,
77 sql: "SELECT * FROM 'iotwireless/uplink'",
78 actions: [{ lambda: { functionArn: host.functionArn } }],
79 });
80
81 yield* Permission(
82 `AWS.IoTWireless.UplinkInvoke(${destination.LogicalId})`,
83 {
84 action: "lambda:InvokeFunction",
85 functionName: host.functionName,
86 principal: "iot.amazonaws.com",
87 sourceArn: rule.ruleArn,
88 },
89 );
90 }),
91 );
92 }
93
94 yield* host.listen(
95 Effect.gen(function* () {
96 return (event: any) => {
97 if (isWirelessUplinkMessage(event)) {
98 return process(Stream.fromArray([event])).pipe(Effect.orDie);
99 }
100 };
101 }),
102 );
103 }) as DestinationEventSourceService;
104 }),
105);