Store raw observations in your backend

from typing import Any, Literal

from pydantic import BaseModel, ConfigDict, Field, field_validator


ALLOWED_STATUSES = {"success", "skipped", "timeout", "error"}


class DeviceSignalEventIn(BaseModel):
    model_config = ConfigDict(extra="forbid")

    action_id: str = Field(max_length=128)
    action_type: str = Field(max_length=64)
    sdk_version: str = Field(max_length=32)
    platform: Literal["android", "ios"]
    app_version: str = Field(default="", max_length=64)
    observations: dict[str, Any]

    @field_validator("observations")
    @classmethod
    def validate_observations(cls, event: dict[str, Any]):
        required = {
            "session_id", "event_type", "schema_version",
            "collected_at", "probes",
        }
        if not required.issubset(event):
            raise ValueError("Invalid RawSignalEvent envelope")
        if event["event_type"] != "device_intel_collection":
            raise ValueError("Unsupported event_type")
        if event["schema_version"] != 1:
            raise ValueError("Unsupported schema_version")
        if not isinstance(event["probes"], dict):
            raise ValueError("probes must be an object")
        for outcome in event["probes"].values():
            if not isinstance(outcome, dict):
                raise ValueError("Invalid probe outcome")
            if outcome.get("status") not in ALLOWED_STATUSES:
                raise ValueError("Unknown probe status")
        return event

from typing import Annotated

from fastapi import Depends, FastAPI, Header, HTTPException, status
from fastapi.responses import JSONResponse

app = FastAPI()


@app.post("/api/v1/device-signal-events", status_code=status.HTTP_201_CREATED)
async def create_device_signal_event(
    body: DeviceSignalEventIn,
    idempotency_key: Annotated[str, Header(alias="Idempotency-Key")],
    account = Depends(require_account),
):
    # Repository stores body.observations unchanged in a JSONB column.
    stored, created = await repository.get_or_create_device_signal_event(
        idempotency_key=idempotency_key,
        account_id=account.id,
        action_id=body.action_id,
        action_type=body.action_type,
        sdk_version=body.sdk_version,
        platform=body.platform,
        payload=body.observations,
    )
    if not created and stored.payload != body.observations:
        raise HTTPException(status_code=409, detail="Idempotency conflict")
    return JSONResponse(
        {"event_id": str(stored.id), "created": created},
        status_code=201 if created else 200,
    )

import express from "express";

const app = express();
app.use(express.json({ limit: "256kb" }));

function assertRawSignalRequest(input) {
  if (!input || typeof input !== "object") throw new Error("Invalid request");
  const event = input.observations;
  if (!event || typeof event !== "object") throw new Error("Invalid observations");
  if (event.event_type !== "device_intel_collection") {
    throw new Error("Unsupported event_type");
  }
  if (event.schema_version !== 1 || !event.probes || typeof event.probes !== "object") {
    throw new Error("Unsupported RawSignalEvent envelope");
  }
  for (const outcome of Object.values(event.probes)) {
    if (!outcome || !["success", "skipped", "timeout", "error"].includes(outcome.status)) {
      throw new Error("Invalid probe outcome");
    }
  }
}

app.post("/api/v1/device-signal-events", requireAccount, async (req, res) => {
  try {
    assertRawSignalRequest(req.body);
  } catch (error) {
    return res.status(400).json({ detail: error.message });
  }

  const idempotencyKey = req.get("Idempotency-Key");
  if (!idempotencyKey) return res.status(400).json({ detail: "Idempotency-Key is required" });

  const result = await db.query(
    `INSERT INTO device_signal_event
      (idempotency_key, account_id, action_id, action_type, sdk_version, platform, payload)
     VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb)
     ON CONFLICT (idempotency_key) DO NOTHING
     RETURNING id`,
    [
      idempotencyKey,
      req.account.id,
      req.body.action_id,
      req.body.action_type,
      req.body.sdk_version,
      req.body.platform,
      JSON.stringify(req.body.observations),
    ],
  );

  if (result.rowCount === 1) {
    return res.status(201).json({ event_id: result.rows[0].id, created: true });
  }
  const existing = await loadByIdempotencyKey(idempotencyKey);
  return res.status(200).json({ event_id: existing.id, created: false });
});

package main

import (
    "encoding/json"
    "net/http"
)

type ingestRequest struct {
    ActionID     string          `json:"action_id"`
    ActionType   string          `json:"action_type"`
    SDKVersion   string          `json:"sdk_version"`
    Platform     string          `json:"platform"`
    AppVersion   string          `json:"app_version"`
    Observations json.RawMessage `json:"observations"`
}

type rawEnvelope struct {
    SessionID     string                     `json:"session_id"`
    EventType     string                     `json:"event_type"`
    SchemaVersion int                        `json:"schema_version"`
    CollectedAt   string                     `json:"collected_at"`
    Probes        map[string]json.RawMessage `json:"probes"`
}

func deviceSignalEvents(w http.ResponseWriter, r *http.Request) {
    if r.Method != http.MethodPost {
        http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
        return
    }
    key := r.Header.Get("Idempotency-Key")
    if key == "" {
        http.Error(w, "Idempotency-Key is required", http.StatusBadRequest)
        return
    }

    r.Body = http.MaxBytesReader(w, r.Body, 256<<10)
    var input ingestRequest
    if err := json.NewDecoder(r.Body).Decode(&input); err != nil {
        http.Error(w, "invalid JSON", http.StatusBadRequest)
        return
    }
    var event rawEnvelope
    if err := json.Unmarshal(input.Observations, &event); err != nil ||
        event.EventType != "device_intel_collection" ||
        event.SchemaVersion != 1 ||
        event.Probes == nil {
        http.Error(w, "invalid RawSignalEvent", http.StatusBadRequest)
        return
    }

    accountID := authenticatedAccountID(r.Context())
    stored, created, err := storeEvent(r.Context(), key, accountID, input)
    if err != nil {
        http.Error(w, "storage failure", http.StatusInternalServerError)
        return
    }
    status := http.StatusOK
    if created {
        status = http.StatusCreated
    }
    w.Header().Set("Content-Type", "application/json")
    w.WriteHeader(status)
    json.NewEncoder(w).Encode(map[string]any{
        "event_id": stored.ID,
        "created": created,
    })
}

func main() {
    http.HandleFunc("/api/v1/device-signal-events", deviceSignalEvents)
    http.ListenAndServe(":8080", nil)
}