Setting the file. One moment. Query Patterns
70
Creating Amazon Aurora Db Cluster With Instances
93
Routing Traffic With Route53 And CloudFront
Resilience Program Design
Creating API Gateway Stage
Position26 of 30scripts/subscription.py
Python·319 lines·14 KB
=
dynamodb.Table(os.environ[
"SUBSCRIBERS_TABLE"
])
13
14# Sparse deprovisioning index markers. On License Deprovisioned the handler writes a constant
15# `deprovisioningPendingFlag` (HASH) and a `deprovisioningExpiry` (RANGE) = event time + the
16# flush window, so the events-stack cleanup Lambda can Query flag='1' AND expiry<=now to
17# finalize expired licenses, while the main-stack sweep/submitter Query flag='1' for the
18# active set. Kept in sync with metering_core.DEPROVISIONING_PENDING_VALUE (subscription.py
19# runs in the events stack and does not import metering_core).
20DEPROVISIONING_PENDING_VALUE = "1"
21DEPROVISIONING_FLUSH_WINDOW = timedelta(hours=1) # AWS Marketplace ~1h post-deprovision window
22
23
24def _mask(value):
25 """Mask a sensitive identifier for logging: keep only the last 4 chars. Buyer account IDs /
26 license ARNs are sensitive and MUST NOT be logged in full (mirrors register.py._mask)."""
27 if not value:
28 return "<none>"
29 s = str(value)
30 return "****" + s[-4:] if len(s) > 4 else "****"
31
32
33def _stamp(update_expr, values):
34 """Append createdAt(once)/updatedAt audit stamps to a SET UpdateExpression (createdAt once, updatedAt every write)."""
35 values = dict(values)
36 values[":now"] = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
37 return (
38 f"{update_expr}, createdAt = if_not_exists(createdAt, :now), updatedAt = :now",
39 values,
40 )
41
42
43def handler(event, context):
44 """Process EventBridge marketplace events from SQS into the unified subscribers table.
45
46 Returns partial batch failures so only the failed message is retried (requires the
47 event source mapping to set FunctionResponseTypes: ReportBatchItemFailures).
48 """
49 batch_item_failures = []
50 for record in event.get("Records", []):
51 try:
52 body = json.loads(record["body"])
53 except (json.JSONDecodeError, KeyError):
54 logger.error("Invalid SQS message body, skipping (not retried)")
55 continue
56
57 try:
58 if "detail-type" in body:
59 _process_event(body)
60 else:
61 logger.warning("Unknown message format, skipping")
62 except Exception:
63 logger.exception("Failed to process record")
64 batch_item_failures.append({"itemIdentifier": record.get("messageId", "")})
65
66 return {"batchItemFailures": batch_item_failures}
67
68
69def _process_event(event):
70 """Route an EventBridge event to the appropriate handler.
71
72 The subscriber row carries TWO independent status fields that this handler owns and
73 MUST NOT conflate:
74
75 * agreementStatus (active | inactive) — the AGREEMENT lifecycle.
76 * subscriptionStatus (active | deprovisioning | inactive) — the LICENSE lifecycle.
77
78 Metering DECISIONS are driven only by the license lifecycle:
79 * `License Updated` -> access granted/refreshed (agreement + subscription active).
80 * `License Deprovisioned`-> access revoked; opens the ~1-hour final-usage flush
81 window (subscriptionStatus = deprovisioning + a
82 deprovisioningExpiry). The events-stack deprovision-cleanup
83 Lambda sets it inactive once the window has elapsed.
84
85 `Purchase Agreement Ended` and `Purchase Agreement Amended` update AGREEMENT status /
86 metadata ONLY — they do NOT stop metering or trigger a flush. `Purchase Agreement
87 Created` is intentionally NOT consumed (it carries no licenseArn and is not needed to
88 meter); sellers who need it add their own rule/target — see
89 references/architecture.md ("Purchase Agreement Created — seller use cases").
90 """
91 detail_type = event.get("detail-type", "")
92 detail = event.get("detail", {})
93 logger.info(f"EventBridge event: {detail_type}")
94
95 if "License Updated" in detail_type:
96 _handle_license_updated(detail)
97 elif "License Deprovisioned" in detail_type:
98 _handle_license_deprovisioned(detail, event.get("time"))
99 elif "Purchase Agreement Ended" in detail_type:
100 _handle_agreement_ended(detail)
101 elif "Purchase Agreement Amended" in detail_type:
102 _handle_agreement_amended(detail)
103 elif "Purchase Agreement Created" in detail_type:
104 # Not consumed by the metering path (no licenseArn; not needed to meter).
105 # Logged only for observability. See references/architecture.md for how sellers
106 # can use this event for other use cases (pre-provisioning, CRM sync, etc.).
107 logger.info("Purchase Agreement Created received; not used by metering path (ignored)")
108
109
110def _handle_license_updated(detail):
111 """`License Updated` — the canonical event that provides the LicenseArn and
112 establishes/refreshes the buyer's access. Upsert the subscriber row (keyed by the
113 real licenseArn) and set BOTH agreementStatus and subscriptionStatus to active.
114 """
115 acceptor_account_id = detail.get("acceptor", {}).get("accountId", "")
116 agreement_id = detail.get("agreement", {}).get("id", "")
117 license_arn = detail.get("license", {}).get("arn", "")
118 product_code = detail.get("product", {}).get("code", "")
119
120 if not acceptor_account_id or not license_arn:
121 logger.error("Missing acceptorAccountId or licenseArn in License Updated event")
122 return
123
124 key = {"licenseArn": license_arn, "customerAWSAccountId": acceptor_account_id}
125 metadata_values = {":pc": product_code, ":aid": agreement_id, ":active": "active"}
126
127 # EventBridge delivery is at-least-once and may be reordered, so a late/duplicate
128 # `License Updated` could arrive AFTER `License Deprovisioned`. Only move
129 # subscriptionStatus to `active` when it is absent or already `active`, so we never
130 # resurrect a `deprovisioning`/`inactive` license and re-open metering for a revoked
131 # buyer. agreementStatus + metadata are always refreshed.
132 try:
133 _mv = dict(metadata_values)
134 _mv[":now"] = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
135 subscribers_table.update_item(
136 Key=key,
137 UpdateExpression=(
138 "SET productCode = :pc, agreementId = :aid, "
139 "agreementStatus = :active, subscriptionStatus = :active, "
140 "createdAt = if_not_exists(createdAt, :now), updatedAt = :now"
141 ),
142 ConditionExpression=(
143 "attribute_not_exists(subscriptionStatus) OR subscriptionStatus = :active"
144 ),
145 ExpressionAttributeValues=_mv,
146 )
147 except subscribers_table.meta.client.exceptions.ConditionalCheckFailedException:
148 # Row is deprovisioning/inactive — refresh agreement metadata but do NOT reactivate.
149 subscribers_table.update_item(
150 Key=key,
151 UpdateExpression=(
152 "SET productCode = :pc, agreementId = :aid, agreementStatus = :active, "
153 "createdAt = if_not_exists(createdAt, :now), updatedAt = :now"
154 ),
155 ExpressionAttributeValues=_mv,
156 )
157 logger.warning(
158 "License Updated arrived for a non-active subscription "
159 f"(account={_mask(acceptor_account_id)} licenseArn={_mask(license_arn)}); refreshed agreement "
160 "metadata but left subscriptionStatus unchanged (not reactivating a revoked license)."
161 )
162 return
163 logger.info(f"License updated (active): account={_mask(acceptor_account_id)} licenseArn={_mask(license_arn)}")
164
165
166def _deprovisioning_expiry(event_time):
167 """ISO-8601 UTC timestamp when the ~1h flush window closes: event time + window.
168
169 Uses the EventBridge event `time` when available (the authoritative moment the license
170 was deprovisioned), else falls back to now. The events-stack cleanup Lambda finalizes a
171 license once this instant has passed.
172 """
173 base = None
174 if event_time:
175 try:
176 base = datetime.fromisoformat(str(event_time).replace("Z", "+00:00"))
177 except ValueError:
178 base = None
179 if base is None:
180 base = datetime.now(timezone.utc)
181 if base.tzinfo is None:
182 base = base.replace(tzinfo=timezone.utc)
183 return (base + DEPROVISIONING_FLUSH_WINDOW).astimezone(timezone.utc).strftime(
184 "%Y-%m-%dT%H:%M:%SZ"
185 )
186
187
188def _handle_license_deprovisioned(detail, event_time=None):
189 """`License Deprovisioned` — access revoked. Opens the ~1-hour final-usage flush window.
190
191 Sets subscriptionStatus = deprovisioning and marks the row in the sparse
192 deprovisioning-pending-index (constant `deprovisioningPendingFlag` HASH + a
193 `deprovisioningExpiry` = event time + ~1h RANGE). During the window the main-stack
194 expedited sweep + submitter flush this license immediately (bypassing MeteringLockHours).
195 Once `deprovisioningExpiry` has passed, the events-stack cleanup Lambda finalizes the
196 license (deprovisioning -> inactive) and removes both markers. Finalization is NOT done by
197 the submitter, because a license may have usage across multiple hours and regions and no
198 single submitter run can know it is fully drained.
199
200 Reference:
201 https://docs.aws.amazon.com/marketplace/latest/userguide/saas-eventbridge-integration.html
202 """
203 acceptor_account_id = detail.get("acceptor", {}).get("accountId", "")
204 license_arn = detail.get("license", {}).get("arn", "")
205
206 if not license_arn:
207 agreement_id = detail.get("agreement", {}).get("id", "")
208 license_arn = _resolve_license_arn(agreement_id)
209
210 if not acceptor_account_id or not license_arn:
211 logger.error("Cannot resolve subscriber for License Deprovisioned event")
212 return
213
214 expiry = _deprovisioning_expiry(event_time)
215 _dep_expr, _dep_vals = _stamp(
216 "SET subscriptionStatus = :s, deprovisioningPendingFlag = :f, deprovisioningExpiry = :exp",
217 {":s": "deprovisioning", ":f": DEPROVISIONING_PENDING_VALUE, ":exp": expiry},
218 )
219 subscribers_table.update_item(
220 Key={
221 "licenseArn": license_arn,
222 "customerAWSAccountId": acceptor_account_id,
223 },
224 UpdateExpression=_dep_expr,
225 ExpressionAttributeValues=_dep_vals,
226 )
227 logger.info(
228 "License deprovisioned (final-usage flush window open until "
229 f"{expiry}): account={_mask(acceptor_account_id)} licenseArn={_mask(license_arn)} "
230 "subscriptionStatus=deprovisioning"
231 )
232
233
234def _handle_agreement_ended(detail):
235 """`Purchase Agreement Ended` — a STATUS update only. Set agreementStatus = inactive.
236
237 This does NOT change subscriptionStatus, stop metering, or trigger a flush: the buyer
238 may still be entitled to usage until the license is deprovisioned. The flush window
239 is opened by `License Deprovisioned`, not by this event.
240 """
241 acceptor_account_id = detail.get("acceptor", {}).get("accountId", "")
242 license_arn = detail.get("license", {}).get("arn", "")
243
244 if not license_arn:
245 agreement_id = detail.get("agreement", {}).get("id", "")
246 license_arn = _resolve_license_arn(agreement_id)
247
248 if not acceptor_account_id or not license_arn:
249 logger.error("Cannot resolve subscriber for Purchase Agreement Ended event")
250 return
251
252 _end_expr, _end_vals = _stamp("SET agreementStatus = :inactive", {":inactive": "inactive"})
253 subscribers_table.update_item(
254 Key={
255 "licenseArn": license_arn,
256 "customerAWSAccountId": acceptor_account_id,
257 },
258 UpdateExpression=_end_expr,
259 ExpressionAttributeValues=_end_vals,
260 )
261 logger.info(
262 "Purchase Agreement Ended (agreement status only; metering unaffected): "
263 f"account={_mask(acceptor_account_id)} licenseArn={_mask(license_arn)} agreementStatus=inactive"
264 )
265
266
267def _handle_agreement_amended(detail):
268 """`Purchase Agreement Amended` — update agreement metadata only (does not change
269 subscriptionStatus or stop metering). The agreement remains active on an amendment.
270 """
271 acceptor_account_id = detail.get("acceptor", {}).get("accountId", "")
272 agreement = detail.get("agreement", {})
273 license_arn = detail.get("license", {}).get("arn", "")
274
275 if not license_arn:
276 agreement_id = agreement.get("id", "")
277 license_arn = _resolve_license_arn(agreement_id)
278
279 if not acceptor_account_id or not license_arn:
280 logger.error("Cannot resolve subscriber for Purchase Agreement Amended event")
281 return
282
283 _amd_expr, _amd_vals = _stamp(
284 "SET endTime = :et, agreementStatus = :active",
285 {":et": agreement.get("endTime", ""), ":active": "active"},
286 )
287 subscribers_table.update_item(
288 Key={
289 "licenseArn": license_arn,
290 "customerAWSAccountId": acceptor_account_id,
291 },
292 UpdateExpression=_amd_expr,
293 ExpressionAttributeValues=_amd_vals,
294 )
295 logger.info(f"Agreement amended: account={_mask(acceptor_account_id)} licenseArn={_mask(license_arn)}")
296
297
298def _resolve_license_arn(agreement_id):
299 """Resolve licenseArn from agreementId via the agreementId GSI (Query, NOT a Scan),
300 used when an Agreement Ended/Amended/Deprovisioned event omits license.arn.
301
302 A genuinely-missing row returns "" (the caller logs and the message is retried/DLQ'd),
303 but permission/config errors (AccessDenied etc.) are NOT swallowed as "not found" —
304 they propagate so the handler's broad except records a batch-item failure (retry ->
305 DLQ alarm) rather than silently leaving the subscriber unchanged.
306 """
307 if not agreement_id:
308 return ""
309
310 response = subscribers_table.query(
311 IndexName="agreementId-index",
312 KeyConditionExpression="agreementId = :agr",
313 ExpressionAttributeValues={":agr": agreement_id},
314 Limit=1,
315 )
316 items = response.get("Items", [])
317 if items:
318 return items[0].get("licenseArn", "")
319 return ""