Einen Serverlosen KI-Barkeeper aufbauen - Teil 2: Gästeregistrierung und Live-Bestellupdates

In Teil 1, habe ich das Fundament für den KI-Cocktailassistenten gebaut. Gäste können die Getränkekarte durchstöbern, ich kann Getränke verwalten und Bilder hochladen, und das System generiert KI-Bildprompts über EventBridge. Aber es gibt ein Problem: Gäste können noch nichts bestellen.

Das bauen wir in diesem Beitrag. Ich brauche drei Dinge, um den Bestellablauf abzuschließen. Erstens müssen Gäste sich registrieren können, ohne vollständige Konten zu erstellen; denken Sie daran, dieses Projekt wurde für eine kleine Hausparty gebaut, und Leute nach E-Mail-Verifizierung und Passwörtern zu fragen, ist zu viel. Zweitens brauchen wir immer noch einen authentifizierten Bestellvorgang, damit Leute ihre Getränkebestellungen absenden und verfolgen können. Drittens möchte ich Echtzeit-Updates, sodass Gäste, wenn ich ein Getränk als fertig markiere, dies sofort auf ihren Handys sehen, anstatt ständig aktualisieren zu müssen.

Also lasst uns bauen!

Warum nicht einfach jeden bestellen lassen?

Die einfachste Lösung wäre, das Bestell-Endpoint komplett öffentlich zu machen, mit einem API-Schlüssel. Jeder kann Bestellungen absenden, ohne sich zu authentifizieren. Aber das erzeugt einige Probleme, die ich nicht haben möchte.

Ohne Benutzeridentifikation kann ich nicht nachverfolgen, wer was bestellt hat. Wenn drei Leute einen Negroni bestellen, wie weiß ich dann, welcher für wen ist, wenn sie fertig sind?

Aus Sicherheitssicht ist ein komplett offenes Endpoint eine Einladung zu Ärger. Jeder im Internet könnte Bestellungen spamen. Selbst bei einer Hausparty möchte ich eine grundlegende Kontrolle.

Die Lösung muss Sicherheit und Komfort balancieren. Gäste sollten keine Hürden überwinden müssen, aber ich brauche genug Kontrolle, um das System nutzbar zu machen.

Gästeregistrierung mit Einladescodes

Für Admins, also mich, verwende ich Amazon Cognito mit vollständiger Authentifizierung: E-Mail, Passwort, JWT-Tokens, alles. Das macht Sinn, wenn ich die Getränkekarte verwalte und alle Bestellungen ansehe. Aber für Partygäste ist das übertrieben.

Ich wollte es so einfach wie möglich machen, wie eine Party-Einladung. Die Idee war, dass Gäste einen Registrierungscode eingeben, ihren Namen angeben und fertig sind.

Wie habe ich also diese Lösung gebaut? Vor der Party generiere ich einen Registrierungscode. Dieser Code hat ein Ablaufdatum, das für den Partytag gültig ist, und eine festgelegte Anzahl von Nutzungen, sodass es möglich ist, einen Code für die erwartete Anzahl von Gästen plus ein paar Extras zu erstellen. Ich kann den Code als scannbaren QR-Code drucken, was die Registrierung noch einfacher macht. Wenn Gäste ankommen, scannen oder geben sie den Code zusammen mit ihrem gewählten Namen ein. Das System validiert den Code, prüft, ob er seine maximale Nutzung erreicht hat oder abgelaufen ist, erstellt ein Benutzerkonto, inkrementiert die Codennutzungsanzahl und gibt JWT-Tokens zurück. Der Gast kann jetzt Bestellungen aufgeben. Aber Moment – JWT-Tokens? Habe ich nicht gesagt, keine JWT-Tokens und Anmeldung? Bleiben Sie ruhig, ich komme zu diesem Teil.

So wurde die Registrierung gebaut.

Das Datenbankschema

Das Registrierungssystem fügt zwei neue Tabellen zum Schema hinzu: app_users für Gäste und registration_codes für die Einladescodes.

CREATE TABLE cocktails.app_users (
    user_key UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    username VARCHAR(100) NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    last_login TIMESTAMP,
    is_active BOOLEAN DEFAULT true,
    metadata TEXT DEFAULT '{}'
);

CREATE TABLE cocktails.registration_codes (
    code UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    created_by VARCHAR(255) NOT NULL,
    expires_at TIMESTAMP NOT NULL,
    max_uses INTEGER NOT NULL,
    use_count INTEGER DEFAULT 0,
    notes TEXT
);

Ich habe die Tabelle app_users anstelle von nur users genannt, da cocktails.users bereits für die Admin-Authentifizierung verwendet wurde. Sie getrennt zu halten macht die Dinge viel klarer.

Das user_key ist eine UUID, die den Gast identifiziert, sodass wir zwei "Sam"s auf der Party haben können.

Registrierungscodes verwenden ebenfalls UUIDs als code-Wert; das macht sie "nicht zu erraten" – na ja, vielleicht ist das übertrieben, aber trotzdem schwer zu erraten. Jeder Code hat eine max_uses-Einschränkung, die bei der Erstellung festgelegt wird, sodass derselbe Code mehrere Gäste registrieren kann. Die use_count verfolgt, wie oft der Code verwendet wurde.

Registrierungscodes erstellen

Ich, als Admin, erstelle Registrierungscodes vor der Party über ein Admin-Endpoint. Die Lambda-Funktion generiert eine UUID und setzt die angegebene Ablaufzeit.

def handler(event, context):
    """Admin-Endpoint - Registrierungscode erstellen."""
    body = json.loads(event["body"])
    expires_at = datetime.fromisoformat(body["expires_at"])
    max_uses = body.get("max_uses", 1)
    
    with get_connection() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """INSERT INTO cocktails.registration_codes
                   (created_by, expires_at, max_uses, notes)
                   VALUES (%s, %s, %s, %s)
                   RETURNING code, expires_at, max_uses""",
                [event["requestContext"]["authorizer"]["userId"], 
                 expires_at, max_uses, body.get("notes", "")]
            )
            result = cur.fetchone()
        conn.commit()
    
    return response(201, {
        "code": str(result["code"]),
        "expires_at": result["expires_at"].isoformat(),
        "max_uses": result["max_uses"]
    })

Der generierte Code sieht ungefähr so aus: f47ac10b-58cc-4372-a567-0e02b2c3d479. Jetzt kann ich ihn den Gästen teilen, wie ich will: per SMS, E-Mail oder als gedruckten QR-Code. Für eine Hausparty mit 20 Gästen könnte ich einen Code mit max_uses: 25 erstellen, um Plus-One-Gäste zu berücksichtigen. Für eine größere Veranstaltung könnte ich mehrere Codes pro Tisch oder Bereich erstellen.

Der Registrierungsablauf

Wenn sich ein Gast registriert, überprüft das System den eingegebenen Code und erstellt das Konto. Der interessante Teil hier ist, dass diese Validierung in zwei verschiedenen Schritten erfolgt. Es beginnt mit einem Lambda-Autorisierer, der läuft, bevor die Anfrage überhaupt die Hauptlogik erreicht, und dann validiert das Endpoint selbst alles erneut. Die Hauptaufgabe des Autorisierers ist es, sicherzustellen, dass der Code echt ist und seine Nutzungsgrenze noch nicht erreicht hat.

def handler(event, context):
    """Registrierungscode-Lambda-Autorisierer."""
    code = event["headers"].get("X-Registration-Code", "")
    if not code:
        return generate_policy("unknown", "Deny", event["methodArn"])
    
    with get_connection() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """SELECT code FROM cocktails.registration_codes
                   WHERE code = %s AND use_count < max_uses AND expires_at > NOW()""",
                [code]
            )
            if not cur.fetchone():
                return generate_policy("unknown", "Deny", event["methodArn"])
    
    return generate_policy(code, "Allow", event["methodArn"],
                          context={"registrationCode": code})

Da der Autorisierer auf der API-Gateway-Ebene arbeitet, erreicht ein falscher Code nicht einmal die Registrierungs-Lambda. Diese Einrichtung ist effizient, da sie verhindert, dass die Registrierungs-Lambda für gefälschte Anmeldungen ausgeführt wird, und so Fluten verhindert. Nach dieser ersten Überprüfung beendet das Registrierungs-Endpoint die Arbeit, indem es den Benutzer in der Tabelle erstellt und die Nutzungsanzahl des Codes erhöht.

def handler(event, context):
    """POST /register - Neuen Gastbenutzer erstellen."""
    code = event["requestContext"]["authorizer"]["registrationCode"]
    username = json.loads(event["body"]).get("username", "").strip()
    
    # Benutzernamen validieren
    if not username or len(username) < 2 or len(username) > 50:
        return response(400, {"error": "Benutzername muss 2-50 Zeichen haben"})
    
    with get_connection() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """INSERT INTO cocktails.app_users (username)
                   VALUES (%s) RETURNING user_key, username""",
                [username]
            )
            user = cur.fetchone()
            
            # Code-Nutzung inkrementieren
            cur.execute(
                "UPDATE cocktails.registration_codes SET use_count = use_count + 1 WHERE code = %s",
                [code]
            )
        conn.commit()
    
    return response(201, {
        "user": {"user_key": str(user["user_key"]), "username": user["username"]},
        "access_token": generate_access_token(user),
        "refresh_token": generate_refresh_token(user)
    })

Das Registrierungs-Endpoint ist in einer wichtigen Hinsicht idempotent. Wenn der Autorisierer besteht, wissen wir, dass der Code gültig ist und seine Grenze noch nicht erreicht hat. Wenn die Lambda mitten im Prozess aufgrund eines Timeouts oder Fehlers scheitert, wird die Nutzungsanzahl nicht inkrementiert, sodass der Gast erneut versuchen kann. Sobald die Lambda die Transaktion commit, erhöht sich die Nutzungsanzahl. Wenn use_count max_uses erreicht, lehnt der Autorisierer zukünftige Registrierungsversuche mit diesem Code ab.

Selbstunterzeichnete JWT-Tokens

Nun zurück zu den JWT-Tokens – ich habe Ihnen gesagt, dass wir darauf zurückkommen.

Nach einer erfolgreichen Registrierung erhalten Gäste JWT-Tokens. Aber im Gegensatz zum Admin-Ablauf, der von Cognito ausgestellte Tokens verwendet, sind Gasttokens selbstunterzeichnet. Ich generiere sie selbst mit RSA-Schlüsselpaaren, die in AWS Secrets Manager gespeichert sind.

Warum selbstunterzeichnet statt Cognito?

Cognito ist ein großartiger Service, aber er ist für langlebige Benutzer konzipiert. Wenn ich ihn hier verwenden würde, müsste ich Benutzerpool-Konten erstellen, Passwörter verwalten und E-Mail-Verifizierungen behandeln. Das ist viel Overhead für einen Gast, der nur für ein paar Stunden da ist.

Selbstunterzeichnete JWTs zu verwenden, gibt mir die Flexibilität, genau zu kontrollieren, was im Token steht und wie lange er aktiv bleibt. Gäste brauchen keine Dinge wie Multi-Faktor-Authentifizierung oder Passwort-Änderungen. Sie brauchen nur einen Weg, um zu beweisen, dass sie sich registriert haben, damit sie für den Abend bestellen können.

Der Haken ist, dass ich die Validierung selbst handhaben muss. Ich muss ein RSA-Schlüsselpaar verwalten und die Verifizierungslogik in einem Lambda-Autorisierer schreiben. Der Prozess ist jedoch nicht so kompliziert, und ich mag es, diese Kontrolle über den gesamten Ablauf zu haben.

Zugriffstoken und Refresh-Tokens erstellen

Die Registrierungs-Lambda generiert zwei Tokens: ein Zugriffstoken und ein Refresh-Token.

def generate_access_token(user: dict) -> str:
    """RS256-unterzeichnetes JWT-Griffzugriffstoken generieren."""
    payload = {
        "token_type": "access",
        "user_key": str(user["user_key"]),
        "username": user["username"],
        "exp": int(time.time()) + (4 * 60 * 60)  # 4 Stunden
    }
    return jwt.encode(payload, get_private_key_from_secrets_manager(), algorithm="RS256")

def generate_refresh_token(user: dict) -> str:
    """Kryptografisch zufälliges Refresh-Token generieren."""
    token = secrets.token_urlsafe(32)
    token_hash = hashlib.sha256(token.encode()).hexdigest()
    
    with get_connection() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """INSERT INTO cocktails.refresh_tokens (user_key, token_hash, expires_at)
                   VALUES (%s, %s, %s)""",
                [user["user_key"], token_hash, datetime.utcnow() + timedelta(days=2)]
            )
        conn.commit()
    return token

Das Zugriffstoken ist ein JWT mit einer Ablaufzeit von 4 Stunden. Vier Stunden reichen für eine Party, sind aber kurz genug, dass gestohlene Tokens nicht ewig gültig bleiben. Die Payload enthält user_key und username, damit nachgeladene Lambda-Funktionen wissen, wer Anfragen stellt.

Das Refresh-Token ist anders. Es ist kein JWT, nur eine kryptografisch zufällige Zeichenfolge. Ich speichere den SHA-256-Hash in der Datenbank zusammen mit seinem Ablaufdatum, 2 Tage. Dies ermöglicht es Gästen, ihre Sitzung über mehrere App-Neustarts hinweg aufrechtzuerhalten, ohne sich neu registrieren zu müssen.

Die Refresh-Token-Tabelle

Refresh-Tokens werden separat von Benutzern gespeichert.

CREATE TABLE cocktails.refresh_tokens (
    token_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    user_key UUID NOT NULL,
    token_hash VARCHAR(64) NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    expires_at TIMESTAMP NOT NULL,
    last_used_at TIMESTAMP,
    is_revoked BOOLEAN DEFAULT false,
    revoked_at TIMESTAMP,
    device_info TEXT
);

Warum den Refresh-Token hashen statt ihn direkt zu speichern? Weil Refresh-Tokens langlebend sind – wenn man 2 Tage langlebend nennen kann. Wenn jemand Zugriff auf die Datenbank erhält, könnte er gespeicherte Tokens verwenden, um Benutzer zu imitieren. Durch das Hashen stellen wir sicher, dass selbst mit Datenbankzugriff der Angreifer die ursprünglichen Tokens nicht rekonstruieren kann.

Das is_revoked-Flag ermöglicht es mir, Tokens bei Bedarf ungültig zu machen. Das device_info-Feld speichert Metadaten wie den Browser-User-Agent, nützlich für Debugging oder um Gästen zu zeigen, wo sie eingeloggt sind.

Refresh-Token-Endpoint

Wenn das Zugriffstoken abläuft, ruft die Frontend-Anwendung den Refresh-Endpoint mit dem Refresh-Token auf.

def handler(event, context):
    """POST /auth/refresh - Tausche Refresh-Token gegen neues Zugriffstoken ein."""
    refresh_token = json.loads(event["body"]).get("refresh_token", "")
    if not refresh_token:
        return response(400, {"error": "refresh_token erforderlich"})
    
    token_hash = hashlib.sha256(refresh_token.encode()).hexdigest()
    
    with get_connection() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """SELECT rt.user_key, u.username FROM cocktails.refresh_tokens rt
                   JOIN cocktails.app_users u ON rt.user_key = u.user_key
                   WHERE rt.token_hash = %s AND NOT rt.is_revoked AND rt.expires_at > NOW()""",
                [token_hash]
            )
            token_record = cur.fetchone()
            if not token_record:
                return response(401, {"error": "Ungültiges oder abgelaufenes Refresh-Token"})
            
            cur.execute(
                "UPDATE cocktails.refresh_tokens SET last_used_at = NOW() WHERE token_hash = %s",
                [token_hash]
            )
        conn.commit()
    
    return response(200, {
        "access_token": generate_access_token({
            "user_key": token_record["user_key"],
            "username": token_record["username"]
        })
    })

Dieses Muster ist ein standardmäßiger OAuth2-ähnlicher Token-Refresh. Das Refresh-Token läuft nie ab, solange es periodisch verwendet wird. Das Zugriffstoken läuft häufig ab, kann aber erneuert werden, ohne sich neu zu authentifizieren.

Der Benutzer-Autorisierer

Bestell-Endpoints erfordern Authentifizierung. Der Benutzer-Autorisierer-Lambda validiert die selbstunterzeichneten JWT-Griffzugriffstokens.

def handler(event, context):
    """Benutzer-JWT-Lambda-Autorisierer - validiert selbstunterzeichnete Tokens."""
    try:
        token = extract_bearer_token(event)
        claims = jwt.decode(
            token,
            get_public_key_from_secrets_manager(),
            algorithms=["RS256"],
            options={"verify_signature": True, "verify_exp": True}
        )
        
        if claims.get("token_type") != "access":
            return generate_policy("unknown", "Deny", event["methodArn"])
        
        return generate_policy(
            claims["user_key"], "Allow", event["methodArn"],
            context={"userId": claims["user_key"], "username": claims["username"]}
        )
    except (jwt.ExpiredSignatureError, jwt.InvalidTokenError):
        return generate_policy("unknown", "Deny", event["methodArn"])

Wie ich sagte, wie Sie sehen können, ist das nicht so komplex. Der Lambda-Autorisierer, der von API Gateway aufgerufen wird, lädt den öffentlichen Schlüssel aus Secrets Manager, verifiziert die Token-Signatur mit der PyJWT-Bibliothek und prüft die Ablaufzeit. Wenn alles passt, extrahiert er das user_key und username und gibt sie an die nachgeladene Lambda-Funktion über den Autorisierer-Kontext weiter.

Nachgeladene Lambda-Funktionen müssen JWT-Tokens nicht selbst parsen. Der Kontext ist unter event["requestContext"]["authorizer"] verfügbar.

Öffentlichen Schlüssel cachen

Eine Optimierung, die erwähnenswert ist, ist das Cachen des öffentlichen Schlüssels. Ihn bei jeder Anfrage von Secrets Manager zu laden, wäre langsam und teuer. Stattdessen cache ich ihn in einer globalen Variable.

_jwt_public_key = None

def get_public_key_from_secrets_manager() -> str:
    """Öffentlichen JWT-Schlüssel von Secrets Manager mit Caching holen."""
    global _jwt_public_key
    
    if _jwt_public_key is not None:
        return _jwt_public_key
    
    client = boto3.client("secretsmanager")
    response = client.get_secret_value(SecretId="drink-assistant/jwt-keys")
    secret = json.loads(response["SecretString"])
    _jwt_public_key = secret["public_key"]
    
    return _jwt_public_key

Lambda-Ausführungsumgebungen werden über Aufrufe wiederverwendet. Der erste Aufruf lädt den Schlüssel und cachet ihn. Nachfolgende Aufrufe in derselben Ausführungsumgebung wiederverwenden den gecachten Schlüssel. Dies reduziert die API-Aufrufe zu Secrets Manager und verbessert die Latenz in vielen Fällen – es ist nicht perfekt, aber sicherlich gut genug.

Bestellungen aufgeben

Nachdem Registrierung und Authentifizierung abgeschlossen sind, können Gäste jetzt Bestellungen aufgeben. Das Bestell-Endpoint ist durch den Benutzer-Autorisierer geschützt.

Die Bestelltabelle

Bestellungen werden in der cocktails.orders-Tabelle gespeichert.

CREATE TABLE cocktails.orders (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    drink_id UUID NOT NULL,
    user_key UUID,
    status VARCHAR(50) NOT NULL DEFAULT 'pending',
    created_at TIMESTAMP DEFAULT NOW(),
    updated_at TIMESTAMP DEFAULT NOW(),
    completed_at TIMESTAMP
);

Das user_key verweist auf app_users.user_key. Das status-Feld verfolgt den Bestelllebenszyklus: pending, preparing, ready oder served. Das completed_at-Zeitstempel wird gesetzt, wenn der Status served erreicht.

Eine Bestellung erstellen

Die Bestell-Lambda ist einfach. Sie überprüft, ob das Getränk verfügbar ist – ich möchte nicht, dass Gäste einen Negroni bestellen, wenn ich kein Campari mehr habe. Dann erstellt sie den Bestell Datensatz und gibt die Bestelldetails zurück.

def handler(event, context):
    """POST /orders - Neue Bestellung für authentifizierten Benutzer erstellen."""
    user_key = event["requestContext"]["authorizer"]["userId"]
    drink_id = json.loads(event["body"]).get("drink_id")
    
    with get_connection() as conn:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT id, name FROM cocktails.drinks WHERE id = %s AND is_active = true",
                [drink_id]
            )
            drink = cur.fetchone()
            if not drink:
                return response(404, {"error": "Getränk nicht gefunden"})
            
            cur.execute(
                """INSERT INTO cocktails.orders (drink_id, user_key, status)
                   VALUES (%s, %s, 'pending') RETURNING id, status, created_at""",
                [drink_id, user_key]
            )
            order = cur.fetchone()
        conn.commit()
    
    publish_order_created(user_key, event["requestContext"]["authorizer"]["username"], 
                         order, drink["name"])
    
    return response(201, {
        "order": {
            "id": str(order["id"]),
            "drink_name": drink["name"],
            "status": order["status"],
            "created_at": order["created_at"].isoformat()
        }
    })

Benutzerbestellungen auflisten

Gäste können auch ihren vollständigen Bestellverlauf abrufen.

def handler(event, context):
    """GET /orders - Bestellungen für authentifizierten Benutzer auflisten."""
    with get_connection() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """SELECT o.id, d.name AS drink_name, o.status, o.created_at
                   FROM cocktails.orders o JOIN cocktails.drinks d ON o.drink_id = d.id
                   WHERE o.user_key = %s ORDER BY o.created_at DESC""",
                [event["requestContext"]["authorizer"]["userId"]]
            )
            return response(200, {"orders": cur.fetchall()})

Die Anfrage verbindet orders mit drinks, um den Getränkenamen in der Antwort einzubeziehen. Gäste sehen "Negroni - ausstehend" statt einer UUID.

Bestellstatus-Updates

Dies ist der Teil, auf den ich sehr gespannt bin und eines der Schlüsselelemente in diesem Teil. Während ich die App baute, erkannte ich, dass es ein potenzielles Problem gab, das eine schlechte Benutzererfahrung schaffen würde. Wenn Sie eine Bestellung aufgeben und keine Möglichkeit haben, den Fortschritt zu verfolgen, kommen Sie einfach zu mir und stören mich mit der Frage nach Ihrem Getränk. Diese Art von manuellem Überprüfen macht den Zweck der App zunichte.

Ich wusste von Anfang an, dass ich eine Form von Live-System brauchte. In dem Moment, in dem ich ein Getränk auf meiner Seite als fertig markiere, sollte sich die App auf dem Handy des Gastes aktualisieren und benachrichtigen. Kein Polling und kein Refresh-Button nötig.

Live-Updates sind normalerweise nur Pub/Sub

Wenn Leute über "Live-Updates" sprechen, meinen sie normalerweise aus technischer Sicht ein Pub/Sub (Publish/Subscribe)-Modell. Genau deshalb habe ich AWS AppSync Events gewählt – es ist ein leistungsstarker Pub/Sub-Service, der es mir ermöglicht, Updates an Benutzer zu pushen.

In einer Standard-Web-App muss der Client Daten anfordern (Anfrage/Antwort). In einem Pub/Sub-Setup abonniert der Client ein bestimmtes Thema, wie orders/{user_id}. Dann, wann immer eine Änderung auf dem Server erfolgt, veröffentlicht der Backend eine Nachricht zu diesem Thema. Der Pub/Sub-Broker (in unserem Fall AppSync Events) kümmert sich um den schweren Teil, herauszufinden, wer zuhört und diese Nachricht sofort an sie zu verteilen.

AppSync Events vs EventBridge

In Teil 1 habe ich Amazon EventBridge für Event-Workflows wie die Generierung von KI-Bildprompts verwendet. EventBridge ist ein großartiger Service, wenn man ereignisgesteuerte Anwendungen baut. Aber EventBridge push nicht Events an Client-Anwendungen.

Um ein Client-Pub/Sub-Setup zu implementieren, muss ich in irgendeiner Form Websockets verwenden. Die einfachste Möglichkeit, die ich gefunden habe, um ein Pub/Sub-Setup über Websockets zu bekommen, ist die Verwendung von AppSync Events – lassen Sie sich nicht durch den AppSync-Teil im Namen täuschen. Dies ist sehr unterschiedlich von der AppSync-API, die eine verwaltete GraphQL-API-Implementierung ist. Es ist ein bisschen bedauerlich, denke ich, dass die Pub/Sub-Implementierung auch unter AppSync gelandet ist.

AppSync Events kümmert sich automatisch um Verbindungen, Abonnements und Broadcasting. Ich veröffentliche einfach Events von Lambda aus, und AppSync leitet sie an abonnierte Clients weiter.

AppSync Events-Konzepte

AppSync Events ist um drei Konzepte aufgebaut.

Kanäle sind wie Themen in einem Pub/Sub-System. Clients abonnieren Kanäle, um Events zu erhalten. In meinem Fall abonniert jeder Benutzer orders/{user_key}, um Updates über seine Bestellungen zu erhalten.

Kanalnamespaces gruppieren verwandte Kanäle. Ich erstelle einen Namespace namens orders, und Kanäle werden dynamisch darin erstellt. Wenn ein Benutzer sich zu orders/abc-123 abonniert, erstellt AppSync diesen Kanal automatisch, wenn er nicht existiert.

Events sind Nachrichten, die an Kanäle veröffentlicht werden. Wenn ich einen Bestellstatus aktualisiere, veröffentliche ich ein Event an den Kanal des Benutzers. AppSync sendet es an alle verbundenen Clients, die diesen Kanal abonniert haben.

Pub/Sub-Designüberblick

Das Pub/Sub-Setup mit AppSync Events sieht ungefähr so aus.

AppSync Events einrichten

Das CloudFormation erstellt eine AppSync Events-API mit Kanalnamespaces.

Resources:
  EventsApi:
    Type: AWS::AppSync::Api
    Properties:
      Name: !Sub "${AWS::StackName}-events-api"
      EventConfig:
        AuthProviders:
          - AuthType: API_KEY
        ConnectionAuthModes:
          - AuthType: API_KEY
        DefaultPublishAuthModes:
          - AuthType: API_KEY
        DefaultSubscribeAuthModes:
          - AuthType: API_KEY

  EventsApiKey:
    Type: AWS::AppSync::ApiKey
    Properties:
      ApiId: !GetAtt EventsApi.ApiId
      Description: API-Schlüssel für AppSync Events
      Expires: 1767225600  # Fernes Ablaufdatum

  OrdersChannelNamespace:
    Type: AWS::AppSync::ChannelNamespace
    Properties:
      ApiId: !GetAtt EventsApi.ApiId
      Name: orders
      SubscribeAuthModes:
        - AuthType: API_KEY
      PublishAuthModes:
        - AuthType: API_KEY

  AdminChannelNamespace:
    Type: AWS::AppSync::ChannelNamespace
    Properties:
      ApiId: !GetAtt EventsApi.ApiId
      Name: admin
      SubscribeAuthModes:
        - AuthType: API_KEY
      PublishAuthModes:
        - AuthType: API_KEY

Ich verwende API-Schlüssel-Authentifizierung für Einfachheit. In der Produktion würden Sie Cognito oder IAM verwenden, um sicherzustellen, dass Benutzer nur ihre eigenen Kanäle abonnieren können. Aber für eine Hausparty ist ein API-Schlüssel, der im Frontend eingebettet ist, ausreichend.

Ich erstelle zwei Kanalnamespaces. Der orders-Namespace ist für benutzerspezifische Bestellupdates. Gäste abonnieren orders/{their_user_key}. Der admin-Namespace ist für Admin-Dashboard-Updates. Ich abonniere admin/new-orders, um zu sehen, wenn neue Bestellungen eingehen.

Events von Lambda veröffentlichen

Wenn sich ein Bestellstatus ändert, veröffentlicht die Lambda ein Event an AppSync.

def publish_order_update(user_key: str, order: dict, drink_name: str):
    """Bestellstatus-Update an den AppSync Events-Kanal des Benutzers veröffentlichen."""
    try:
        appsync_client.post(
            action="publish",
            apiId=os.environ["APPSYNC_API_ID"],
            payload=json.dumps({
                "channel": f"orders/{user_key}",
                "events": [json.dumps({
                    "type": "ORDER_STATUS_CHANGED",
                    "order_id": str(order["id"]),
                    "drink_name": drink_name,
                    "status": order["status"],
                    "updated_at": order["updated_at"].isoformat()
                })]
            })
        )
    except Exception as e:
        logger.error(f"Fehler beim Veröffentlichen: {e}")

Das Veröffentlichen ist Feuer und vergessen. Wenn es scheitert, protokolliere ich den Fehler, aber ich lasse die gesamte Bestellaktualisierung nicht scheitern. Der Bestellstatus wird unabhängig in der Datenbank gespeichert. Im schlimmsten Fall erhält der Gast keine Echtzeit-Benachrichtigung und muss manuell aktualisieren.

Neue Bestell-Events an das Admin-Dashboard veröffentlichen

Wenn ein Gast eine Bestellung erstellt, veröffentliche ich auch am Admin-Kanal, damit ich sofort Bescheid weiß.

def publish_order_created(user_key: str, username: str, order: dict, drink_name: str):
    """Neues Bestell-Event an den Admin-Kanal veröffentlichen."""
    try:
        appsync_client.post(
            action="publish",
            apiId=os.environ["APPSYNC_API_ID"],
            payload=json.dumps({
                "channel": "admin/new-orders",
                "events": [json.dumps({
                    "type": "NEW_ORDER",
                    "order_id": str(order["id"]),
                    "user_key": user_key,
                    "username": username,
                    "drink_name": drink_name,
                    "status": "pending",
                    "created_at": order["created_at"].isoformat()
                })]
            })
        )
    except Exception as e:
        logger.error(f"Fehler beim Veröffentlichen: {e}")

Jetzt erhalten sowohl die Gast-App als auch das Admin-Dashboard Echtzeit-Updates. Gäste sehen, wie sich ihr Bestellstatus ändert. Ich sehe neue Bestellungen sofort.

Vom Frontend abonnieren

Im Frontend verwende ich die AWS Amplify-Bibliothek, um AppSync Events-Kanäle zu abonnieren.

Zuerst Amplify mit den AppSync Events-API-Details konfigurieren.

import { Amplify } from 'aws-amplify';

Amplify.configure({
  API: {
    Events: {
      endpoint: 'https://abc123.appsync-api.us-east-1.amazonaws.com/event',
      region: 'us-east-1',
      defaultAuthMode: 'apiKey',
      apiKey: 'da2-xxxxxxxxxxxxxxxxxxxxxxxxxx'
    }
  }
});

Dann abonnieren Sie den Bestellkanal des Benutzers.

import { events } from '@aws-amplify/api';

function subscribeToOrderUpdates(userKey: string) {
  const channel = events.channel(`orders/${userKey}`);
  
  const subscription = channel.subscribe({
    next: (event) => {
      console.log('Bestell-Update empfangen:', event);
      
      // UI basierend auf Event-Typ aktualisieren
      if (event.type === 'ORDER_STATUS_CHANGED') {
        updateOrderInUI(event.order_id, event.status);
        
        if (event.status === 'ready') {
          showNotification(`Ihr ${event.drink_name} ist fertig!`);
        }
      }
    },
    error: (error) => {
      console.error('AppSync Events-Fehler:', error);
    }
  });
  
  // Abmeldungsfunktion zurückgeben
  return () => subscription.unsubscribe();
}

Das Kanalabonnement verwendet Websockets im Hintergrund. Wenn die Lambda ein Event veröffentlicht, push AppSync es an alle verbundenen Clients, die diesen Kanal abonniert haben. Die next-Callback wird sofort ausgelöst, wenn Events eintreffen.

React-Hook für Echtzeitbestellungen

Ich habe die Abonnementlogik in einen React-Hook gewickelt, der den Verbindungslebenszyklus verwaltet.

import { useEffect, useState } from 'react';
import { events } from '@aws-amplify/api';

interface Order {
  id: string;
  drink_id: string;
  drink_name: string;
  status: 'pending' | 'preparing' | 'ready' | 'served';
  created_at: string;
  updated_at: string;
}

export function useRealtimeOrders(userKey: string) {
  const [orders, setOrders] = useState<Order[]>([]);
  const [connected, setConnected] = useState(false);

  useEffect(() => {
    if (!userKey) return;

    const channel = events.channel(`orders/${userKey}`);

    const subscription = channel.subscribe({
      next: (event) => {
        if (event.type === 'ORDER_STATUS_CHANGED') {
          // Den spezifischen Auftrag im Zustand aktualisieren
          setOrders((prevOrders) => 
            prevOrders.map((order) =>
              order.id === event.order_id
                ? { ...order, status: event.status, updated_at: event.updated_at }
                : order
            )
          );

          // Benachrichtigung für fertige Bestellungen anzeigen
          if (event.status === 'ready') {
            new Notification(`${event.drink_name} ist fertig!`);
          }
        }
      },
      error: (error) => {
        console.error('Abonnementfehler:', error);
        setConnected(false);
      }
    });

    setConnected(true);

    // Aufräumarbeiten beimUnmounten
    return () => {
      subscription.unsubscribe();
      setConnected(false);
    };
  }, [userKey]);

  return { orders, setOrders, connected };
}

Die Verwendung des Hooks in einem Component ist einfach.

function OrdersPage({ userKey }: { userKey: string }) {
  const { orders, setOrders, connected } = useRealtimeOrders(userKey);

  // Initiale Bestellungen beim Mount abrufen
  useEffect(() => {
    async function loadOrders() {
      const response = await fetch('/api/orders', {
        headers: { 'Authorization': `Bearer ${accessToken}` }
      });
      const data = await response.json();
      setOrders(data.orders);
    }
    loadOrders();
  }, [userKey, setOrders]);

  return (
    <div>
      <div className="connection-status">
        {connected ? '🟢 Live' : '🔴 Offline'}
      </div>
      {orders.map(order => (
        <OrderCard key={order.id} order={order} />
      ))}
    </div>
  );
}

Der Hook verwaltet den Abonnementslebenszyklus automatisch. Wenn das Component montiert wird, abonniert es. Wenn es abgemontet wird, trennt es sich. Der connected-Zustand zeigt Gästen an, ob sie Live-Updates erhalten.

Admin-Dashboard-Abonnement

Das Admin-Dashboard abonniert den admin/new-orders-Kanal.

function AdminDashboard() {
  const [pendingOrders, setPendingOrders] = useState<Order[]>([]);

  useEffect(() => {
    const subscription = events.channel('admin/new-orders').subscribe({
      next: (event) => {
        if (event.type === 'NEW_ORDER') {
          setPendingOrders(prev => [{
            id: event.order_id,
            drink_name: event.drink_name,
            username: event.username,
            status: 'pending',
            created_at: event.created_at
          }, ...prev]);
        }
      }
    });
    return () => subscription.unsubscribe();
  }, []);

  // Pendende Bestellungen rendern...
}

Neue Bestellungen erscheinen sofort, ohne Polling.

Wiederherstellung und Zuverlässigkeit

AppSync Events handelt Wiederherstellung automatisch. Wenn ein Client die Verbindung verliert, versucht AppSync, sich wieder anzumelden. Wenn die Verbindung wiederhergestellt ist, wird das Abonnement fortgesetzt.

Aber hier ist eine wichtige Einschränkung: Events, die veröffentlicht wurden, während der Client getrennt war, gehen verloren. AppSync Events ist flüchtig. Es wartet keine Nachrichten oder spielt verpasste Events nicht ab. Aus diesem Grund behandle ich die Datenbank als Quelle der Wahrheit und AppSync als Geschwindigkeits-Ebene.

Die Echtzeit-Updates sind da, um die UI schnell und reaktionsschnell zu machen, während der Gast aktiv die App nutzt. Wenn die Verbindung abbricht, behandelt die Amplify-Bibliothek die Wiederherstellung der Verbindung automatisch. Da wir jedoch wissen, dass Nachrichten nicht gewartet werden, muss die App eine Möglichkeit haben, den Verbindungsstatus zu erkennen.

Immer wenn die App sich von einem Offline-Zustand erholt oder der Gast manuell aktualisieren, sollte das Frontend eine schnelle GET /orders-Anfrage an die REST-API senden. Dies stellt sicher, dass selbst wenn sie eine Live-Benachrichtigung verpasst haben, der Bildschirm immer mit dem tatsächlichen Zustand der Datenbank synchronisiert ist.

Es ist ein "Bestes aus beiden Welten"-Ansatz. Sie erhalten den sofortigen "Wow"-Effekt einer Live-Update, wenn alles perfekt funktioniert, aber Sie haben die Zuverlässigkeit einer traditionellen Datenbank, auf die Sie zurückgreifen können, wenn das WLAN auf der Party etwas unzuverlässig wird.

Was kommt als Nächstes

Das Fundament und der Bestellablauf sind abgeschlossen. Gäste können Getränke durchstöbern, sich mit Einladescodes registrieren, Bestellungen aufgeben und Echtzeit-Updates erhalten, wenn ihre Getränke fertig sind. Ich kann die Speisekarte verwalten, Bestellungen anzeigen und den Bestellstatus aktualisieren. Alles funktioniert, und es funktioniert gut.

Aber das KI-Teil fehlt noch. In Teil 3 füge ich die Funktionen hinzu, die dies zu einem KI-Barkeeper machen, anstatt nur zu einer digitalen Speisekarte. KI-gestützte Getränkeempfehlungen basierend auf Geschmackspräferenzen. Konversationelle Bestellung über eine Chat-Oberfläche. Bildgenerierung für benutzerdefinierte Getränkefotos mit Amazon Bedrock Nova Canvas. Alles integriert mit dem Fundament, das wir gebaut haben.

Teil 3 kommt bald. Folgen Sie mir auf LinkedIn, damit Sie es nicht verpassen.

Letzte Worte

Echtzeit-Updates waren einmal schwer. Websockets, Verbindungsverwaltung, Broadcasting, Skalierung. Jetzt kümmert sich AppSync Events um alles. Sie veröffentlichen von Lambda aus, Clients abonnieren, und AppSync leitet Events weiter. So einfach ist das.

Das selbstunterzeichnete JWT-Muster ist ebenso einfach. Erstellen Sie ein RSA-Schlüsselpaar, unterschreiben Sie Tokens in einer Lambda, überprüfen Sie sie in einer anderen. Kein Cognito-Overhead für temporäre Benutzer, nur einfache Kryptografie.

Das Ergebnis ist ein vollständiges Bestellsystem, das sich schnell und reaktionsschnell anfühlt. Gäste registrieren sich in Sekunden, geben Bestellungen mit einer Berührung auf und werden benachrichtigt, sobald ihr Getränk fertig ist. Kein Polling, kein manuelles Aktualisieren, keine Reibung.

Der vollständige Code ist auf GitHub. Schauen Sie sich meine anderen Beiträge auf jimmydqv.com für weitere Serverless-Muster an.

Jetzt loslegen!