feat(login_events): Redis pub/sub on successful logins

Publishes `{user_id, provider}` JSON to the `placeos/auth/login`
Redis channel after every successful login — same channel the Ruby
service used, so downstream subscribers (cache invalidation, audit
logging, …) keep working unchanged.

Wired in two places:
  * Sessions#signin (provider = "internal")
  * ProviderCallbacks#callback (provider = oauth_user.provider)

Errors are caught + logged at warn — a flaky Redis must never block
a successful login. The publisher is exposed as a class_property
Proc so tests (and any future in-process subscriber) can swap in a
recording double; that's how the new 2 specs assert the call without
spinning up a Redis subscriber.

The legacy Ruby `before_signup` / `after_login` Rails-initialiser
hooks are not ported: auth.cr is a standalone binary with no obvious
in-process consumer for them. Easy to add a callable class_property
when a concrete need shows up.

41/41 specs pass (./test).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: stakach

spec/controllers/login_events_spec.cr

@@ -0,0 +1,106 @@
+require "webmock"
+require "../helper"
+
+# Verifies that successful logins publish a `{user_id, provider}` event
+# through `LoginEvents`. We don't go through real Redis here — we swap
+# `LoginEvents.publisher` for a recording Proc. This keeps the spec
+# independent of the Redis stub in `docker-compose.yml` and proves the
+# wire-up (which is the part that can rot under refactors); the actual
+# `redis` PUBLISH call is a single line covered by the shard's own
+# spec suite.
+module PlaceOS::Auth
+  describe LoginEvents do
+    # Save/restore the publisher on every example so a failure half-way
+    # through doesn't leak a recording stub into the next test.
+    original_publisher = LoginEvents.publisher
+    ::Spec.before_each { LoginEvents.publisher = original_publisher }
+    ::Spec.after_each { LoginEvents.publisher = original_publisher }
+
+    describe "publish" do
+      it "fires `{user_id, internal}` after a successful POST /auth/signin" do
+        calls = [] of Tuple(String, String)
+        LoginEvents.publisher = ->(uid : String, provider : String) {
+          calls << {uid, provider}
+          nil
+        }
+
+        authority = ::PlaceOS::Model::Authority.find_by_domain("localhost").not_nil!
+        password = "ok-password-1234"
+        user = ::PlaceOS::Model::Generator.user(authority)
+        user.password = password
+        user.save!
+
+        body = {email: user.email.to_s, password: password}.to_json
+        headers = HTTP::Headers{
+          "Host"         => "localhost",
+          "Content-Type" => "application/json",
+        }
+        result = client.post("/auth/signin", headers: headers, body: body)
+        result.status_code.should eq 202
+
+        calls.size.should eq 1
+        calls.first[0].should eq user.id
+        calls.first[1].should eq "internal"
+      ensure
+        user.try &.destroy
+      end
+
+      it "fires `{user_id, oauth2}` after a successful OAuth callback" do
+        calls = [] of Tuple(String, String)
+        LoginEvents.publisher = ->(uid : String, provider : String) {
+          calls << {uid, provider}
+          nil
+        }
+
+        authority = ::PlaceOS::Model::Authority.find_by_domain("localhost").not_nil!
+        strat = ::PlaceOS::Model::OAuthAuthentication.new
+        strat.name = "login-event-#{Random.rand(99999)}"
+        strat.client_id = "test"
+        strat.client_secret = "test"
+        strat.site = "https://idp.example.test"
+        strat.authorize_url = "/authorize"
+        strat.token_url = "/token"
+        strat.auth_scheme = "request_body"
+        strat.token_method = "post"
+        strat.scope = "openid email"
+        strat.raw_info_url = "https://idp.example.test/userinfo"
+        strat.info_mappings = {"uid" => "id", "email" => "email", "name" => "name"}
+        strat.authority_id = authority.id
+        strat.save!
+
+        WebMock.reset
+        WebMock.allow_net_connect = false
+        WebMock.stub(:post, "https://idp.example.test/token").to_return(
+          status: 200,
+          headers: HTTP::Headers{"Content-Type" => "application/json"},
+          body: {access_token: "x", token_type: "Bearer", expires_in: 3600}.to_json,
+        )
+        WebMock.stub(:get, "https://idp.example.test/userinfo").to_return(
+          status: 200,
+          headers: HTTP::Headers{"Content-Type" => "application/json"},
+          body: {id: "uid-#{Random.rand(99999)}", email: "bob-#{Random.rand(99999)}@localhost", name: "Bob"}.to_json,
+        )
+
+        kickoff = client.get(
+          "/auth/oauth2?id=#{URI.encode_www_form(strat.id.as(String))}",
+          headers: HTTP::Headers{"Host" => "localhost"},
+        )
+        kickoff.status_code.should eq 303
+        cookie = kickoff.headers["Set-Cookie"].split(';', 2).first.strip
+        state = URI::Params.parse(kickoff.headers["Location"].split('?', 2).last)["state"]
+
+        callback = client.get(
+          "/auth/oauth2/callback?id=#{URI.encode_www_form(strat.id.as(String))}&code=test-code&state=#{state}",
+          headers: HTTP::Headers{"Host" => "localhost", "Cookie" => cookie},
+        )
+        callback.status_code.should eq 303
+
+        calls.size.should eq 1
+        calls.first[1].should eq "oauth2"
+      ensure
+        strat.try &.destroy
+        WebMock.reset
+      end
+    end
+  end
+end

src/placeos-auth.cr

@@ -5,6 +5,7 @@ require "./constants"
 require "./placeos-auth/error"
 require "./placeos-auth/authly_adapter"
 require "./placeos-auth/external_providers"
+require "./placeos-auth/login_events"
 require "./placeos-auth/controllers/application"
 require "./placeos-auth/controllers/*"
 

src/placeos-auth/controllers/provider_callbacks.cr

@@ -99,6 +99,8 @@ module PlaceOS::Auth
       remove_session
       new_session(result.user)
 
+      LoginEvents.publish(result.user, oauth_user.provider)
+
       target = consume_continue || "/"
       redirect_to target.gsub(' ', "%20"), :see_other
     end

src/placeos-auth/controllers/sessions.cr

@@ -56,8 +56,7 @@ module PlaceOS::Auth
       remove_session
       new_session(user)
 
-      # TODO(phase 6): publish `{user_id, provider: "internal"}` to the
-      # `placeos/auth/login` Redis channel.
+      LoginEvents.publish(user, "internal")
 
       if (continue = body.continue.presence)
         target = sanitize_continue(continue) || "/"

src/placeos-auth/login_events.cr

@@ -0,0 +1,68 @@
+require "json"
+require "redis"
+
+module PlaceOS::Auth
+  # Publishes successful logins onto a Redis pub/sub channel so the
+  # rest of the PlaceOS stack can observe authentication events
+  # (cache invalidation, audit logs, etc.).
+  #
+  # Mirrors the legacy Ruby service's `placeos/auth/login` channel —
+  # downstream subscribers expect the exact same payload shape:
+  # `{"user_id": "...", "provider": "..."}`.
+  module LoginEvents
+    Log = ::PlaceOS::Auth::Log.for(self)
+
+    # Indirection so tests can swap in a recording double without
+    # spinning up a Redis subscriber. Production code calls
+    # `LoginEvents.publish` which delegates here; tests reassign
+    # `LoginEvents.publisher` to a Proc that captures invocations.
+    class_property publisher : Proc(String, String, Nil) = ->(user_id : String, provider : String) {
+      publish_to_redis(user_id, provider)
+    }
+
+    # Single shared connection. `redis` is thread-safe per docs but we
+    # only `publish` (no blocking subscribers) so a single instance is
+    # fine. Lazy-init so tests don't pay the connection cost unless
+    # something actually publishes.
+    @@redis : ::Redis? = nil
+    @@redis_mutex = Mutex.new
+
+    # Fire-and-forget. Errors are logged at `warn` and swallowed —
+    # a flaky Redis must never block a successful login.
+    def self.publish(user : ::PlaceOS::Model::User, provider : String) : Nil
+      uid = user.id
+      return if uid.nil?
+      publisher.call(uid, provider)
+    end
+
+    # :nodoc:
+    def self.publish_to_redis(user_id : String, provider : String) : Nil
+      redis_url = REDIS_URL
+      return if redis_url.nil? || redis_url.empty?
+
+      payload = {user_id: user_id, provider: provider}.to_json
+      redis = ensure_connection(redis_url)
+      return if redis.nil?
+      redis.publish(LOGIN_EVENTS_CHANNEL, payload)
+    rescue ex
+      Log.warn(exception: ex) { {action: "login_events.publish", message: "ignoring failure"} }
+    end
+
+    # Resets the singleton — useful for tests that want a clean state
+    # or to recover after a Redis-side restart.
+    def self.reset_connection : Nil
+      @@redis_mutex.synchronize do
+        @@redis.try &.close
+        @@redis = nil
+      end
+    end
+
+    private def self.ensure_connection(url : String) : ::Redis?
+      @@redis_mutex.synchronize do
+        existing = @@redis
+        return existing if existing
+        @@redis = ::Redis.new(url: url)
+      end
+    end
+  end
+end

tasks/todo.md

@@ -77,9 +77,12 @@ Granular checklist that mirrors `PLAN.md`. Check off as we go; capture correctio
 - [x] Full SAMLResponse signature-validation round-trip deferred — `multi_auth_saml`'s own spec suite covers SAML XML parsing; the auth.cr-specific bits (user mapping, state check) are exercised through the shared code path by `provider_callbacks_spec.cr`
 
 ## Phase 6 — Extensions
-- [ ] Redis login event publisher (`placeos/auth/login`)
-- [ ] Crystal-native `before_signup` / `after_login` hooks
-- [ ] Finalise API key validator + X-API-Key header support
+- [x] Redis login event publisher (`LoginEvents.publish` → `placeos/auth/login`)
+- [x] Wired into `Sessions#signin` (provider="internal") and `ProviderCallbacks#callback` (provider=oauth_user.provider)
+- [x] Test-swappable publisher (`LoginEvents.publisher` class_property)
+- [x] Specs: signin → "internal" event, OAuth callback → provider name event (2 specs)
+- [ ] Crystal-native `before_signup` / `after_login` hooks — deferred. The Ruby version was a Rails-initialiser plug-in pattern; auth.cr is a standalone binary so there's no obvious consumer. Add when a concrete need arises.
+- [x] X-API-Key header already supported via `Utils::CurrentUser#extract_api_key` (Phase 1)
 
 ## Phase 7 — Cutover prep
 - [ ] Wire-format diff vs Ruby service (cookies, JWT, response codes/headers)