Lexnis Residential

A residential estate needs a way for residents to raise an emergency and have an on-duty guard actually respond, with a record of what happened afterwards. Before this, the "system" was a phone call or a WhatsApp group — no way to know which guard was assigned, whether they were on their way, when they arrived, or what the incident actually was. Nothing was logged, so there was no accountability trail when a resident asked "what happened to my alert last Tuesday?" Guards, residents and the estate admin were all working off memory and screenshots.

Three clients consume one backend — a customer app, an officer app (both Flutter), and an admin panel (Next.js) — so the API had to be the single source of truth and every state change had to reach all three in real time. The target host is serverless (NeonDB Postgres over asyncpg), which rules out anything that assumes a long-lived process owns the database. Guards and residents are on phones with flaky connectivity, so "real-time" could not mean "only works while the socket is open." And this is an emergency product — a dropped or ambiguous SOS is a safety failure, not a bug ticket.

The service is a thin-route / fat-controller FastAPI app: routers do auth, validation and HTTP status; all business logic lives in controllers (officer_controller, customer_controller, admin_controller, …). Everything is async end to end against Postgres via asyncpg, because the same request that writes an SOS also fans it out to live sockets and fires a push — blocking I/O anywhere in that path stalls the event loop.

The two decisions that shaped everything: one unified messages table and a dual real-time delivery path. Rather than separate tables for group chat, resident↔admin queries, SOS channels and internal admin chat, there is a single messages table keyed by channel_type + channel_id. That means one query path, one media pipeline, one pagination implementation for every kind of conversation — at the cost of the table meaning nothing without its discriminator, which I accepted. For delivery, live updates go over an in-process WebSocket ConnectionManager (an asyncio.Lock-guarded dict of channel → sockets), and the *same* events also go out as Firebase Cloud Messaging pushes. WebSocket handles "resident is staring at the screen"; FCM handles "resident's phone is in their pocket." The push path is deliberately fire-and-forget — it logs and swallows every failure and never raises, because a Firebase outage must not block an SOS from being written.

Media (chat images, incident attachments) is stored in S3 by key, not by URL: the DB holds media_key and a presigned URL is generated fresh on every read with a 1-hour expiry, so no long-lived public links leak.

It started as a straightforward CRUD backend — users, areas, shifts, a chat table — and the SOS was going to be "insert a row, notify a guard." That lasted until the first real conversation about accountability, when it became obvious that an alert without a lifecycle is just a louder phone call. The migration history is honest about this: the very first migration adds ETA/arrived/resolved timing columns to a table that already existed, because I hadn't seen the state machine coming.

What I got wrong early was trusting the ORM's defaults. I let SQLModel generate native Postgres enums and moved on, and it worked in every test until a filtered query returned nothing in a real database and I couldn't see why. The DB had SOS_CSV, the query sent sos_csv, and nothing errored — it just quietly matched zero rows. Tracking that down, and then discovering Postgres won't let you remove an enum value or add one inside a transaction, was the week that taught me to stop reaching for native enums.

The moment it clicked was the customer-approval step. Once I stopped treating "resolved" as a boolean the officer flips and modeled it as a claim the resident confirms — with its own rejection and consent branches — the whole thing made sense. The rating system, the incident report, the reference numbers all hung off that spine cleanly, because there was finally a real object with a real end state to attach them to.

If I did it again I'd design the SOS state machine and the reference-number scheme on day one instead of retrofitting them across a dozen migrations, and I'd use VARCHAR for status columns from the start. I'd also decide the WebSocket-plus-FCM dual delivery contract up front rather than discovering per-feature that a socket broadcast alone never reaches a backgrounded phone.

The one thing I'd keep exactly as-is is the fire-and-forget push layer. Every time Firebase misbehaved in development, the SOS still wrote, still broadcast, still health-checked green — and that separation is the difference between a degraded feature and a down product.

Shipped as the live backend for the Lexins platform, serving a Flutter customer app, a Flutter officer app and a Next.js admin panel from one API. What's in the repo: 17 tables, 15 Alembic migrations, ~165 REST and WebSocket endpoints, a full SOS lifecycle from raise through officer response to resident-confirmed resolution with an attached incident report, unified real-time chat, and dual WebSocket + FCM delivery. It is backed by 730 test functions across 42 test modules covering auth, admin analytics, the SOS flow, ratings, notifications and media handling. As of July 2026 the client-feedback round (daily occurrence logs, officer-initiated SOS, map tracking, multi-officer dispatch) is planned and in progress.

Lexins SOS is a comprehensive community management system for residential estates. The backend is a fully async FastAPI (Python 3.13) + WebSocket server with 60+ REST endpoints and 3 WebSocket channels, serving Admin, Customer, and Security Officer roles via JWT auth with refresh token rotation.

The SOS emergency system broadcasts real-time alerts to active officers via WebSocket and tracks the full lifecycle: SENT → RESPONDED → PENDING_APPROVAL → RESOLVED with ETA, arrival time, resolution time, and a post-SOS officer rating. AWS S3 CSV exports use boto3 presigned URLs with a graceful BYTEA fallback. The CI/CD pipeline runs: ruff lint → pytest (200+ tests, Postgres 16 container) → docker buildx → push to AWS ECR → SSH to EC2 → migrate → health-check rollback.