diff --git a/spec/app/listen_spec.cr b/spec/app/listen_spec.cr index 0749f84..7b8b4e3 100644 --- a/spec/app/listen_spec.cr +++ b/spec/app/listen_spec.cr @@ -1,5 +1,6 @@ require "../spec_helper" require "http/client" +require "socket" private def wait_for_port(host : String, port : Int32, timeout : Time::Span = 5.seconds) deadline = Time.instant + timeout @@ -182,4 +183,74 @@ describe "App#close and graceful shutdown" do request_finished.receive end end + + describe "App#listen with Unix Socket" do + it "binds to a unix socket and sets ctx.provider to local" do + app = Alumna::App.new + app.use "/provider", Alumna.memory(Alumna::Schema.new) { + before do |ctx| + ctx.result = {"provider" => ctx.provider} of String => Alumna::AnyData + nil + end + } + + port = 34572 + unix_path = "/tmp/alumna_test_#{Random::Secure.hex(4)}.sock" + + spawn do + app.listen(port, host: "127.0.0.1", unix_socket: unix_path, trap_signals: false) + end + + wait_for_port("127.0.0.1", port) + + begin + # 1. Test TCP -> "rest" + res_tcp = HTTP::Client.get("http://127.0.0.1:#{port}/provider") + res_tcp.status_code.should eq(200) + JSON.parse(res_tcp.body)["provider"].as_s.should eq("rest") + + # 2. Test Unix -> "local" + socket = UNIXSocket.new(unix_path) + socket << "GET /provider HTTP/1.1\r\nHost: localhost\r\n\r\n" + res_unix = HTTP::Client::Response.from_io(socket) + res_unix.status_code.should eq(200) + JSON.parse(res_unix.body)["provider"].as_s.should eq("local") + socket.close + ensure + app.close + File.delete?(unix_path) + end + end + + it "binds solely to a unix socket when port is nil" do + app = Alumna::App.new + app.use "/only-unix", Alumna::MemoryAdapter.new(Alumna::Schema.new) + + unix_path = "/tmp/alumna_test_only_#{Random::Secure.hex(4)}.sock" + listen_done = Channel(Nil).new + + spawn do + app.listen(port: nil, unix_socket: unix_path, trap_signals: false) + listen_done.send(nil) + end + + deadline = Time.instant + 5.seconds + while !File.exists?(unix_path) + raise "Socket not created" if Time.instant > deadline + sleep 0.05.seconds + end + + begin + socket = UNIXSocket.new(unix_path) + socket << "GET /only-unix HTTP/1.1\r\nHost: localhost\r\n\r\n" + res_unix = HTTP::Client::Response.from_io(socket) + res_unix.status_code.should eq(200) + socket.close + ensure + app.close + listen_done.receive + File.delete?(unix_path) + end + end + end end diff --git a/spec/integration/app_inter_service_comm_spec.cr b/spec/integration/app_inter_service_comm_spec.cr new file mode 100644 index 0000000..7816c74 --- /dev/null +++ b/spec/integration/app_inter_service_comm_spec.cr @@ -0,0 +1,130 @@ +require "../spec_helper" +require "../../src/testing" + +# --- 1. Schemas --- +UserSchema = Alumna::Schema.new + .str("name") + +PostSchema = Alumna::Schema.new + .str("author_id") + .str("content") + +AuditSchema = Alumna::Schema.new + .str("action") + .str("resource_id") + +# --- 2. Rules --- +AuthenticateGlobal = Alumna::Rule.new do |ctx| + # If it's an internal call from another service, trust it automatically. + next nil if ctx.provider == "internal" || ctx.provider == "local" + + token = ctx.headers["authorization"]? + if token == "Bearer secret-admin" + # Put the user id in the store. Internal calls will inherit this! + ctx.store["current_user_id"] = "1" + nil + else + Alumna::ServiceError.unauthorized + end +end + +ValidateAuthorExists = Alumna::Rule.new do |ctx| + author_id = ctx.data_str?("author_id") + + begin + # INTERNAL CALL 1: Fetch from the users service + ctx.call("/users", :get, id: author_id) + nil + rescue ex + # If the call fails (e.g., 404), bubble it up as a validation error + Alumna::ServiceError.unprocessable("Validation failed", {"author_id" => "Author does not exist"} of String => Alumna::AnyData) + end +end + +CreateAuditLog = Alumna::Rule.new do |ctx| + # INTERNAL CALL 2: Fire-and-forget audit log creation + ctx.call("/audit", :create, { + "action" => "Created Post", + "resource_id" => ctx.result.as(Hash(String, Alumna::AnyData))["id"], + } of String => Alumna::AnyData) + + nil +end + +# --- 3. App Setup --- +APP_COMM = Alumna::App.new.tap do |app| + app.before AuthenticateGlobal + + # Notice how we can use the cleaner syntax here now! + app.use "/users", Alumna.memory(UserSchema) + app.use "/audit", Alumna.memory(AuditSchema) + + app.use "/posts", Alumna.memory(PostSchema) { + before ValidateAuthorExists, on: :create + after CreateAuditLog, on: :create + } +end + +# --- 4. Specs --- +describe "Inter-Service Communication (ctx.call)" do + client = Alumna::Testing::AppClient.new(APP_COMM) + client.default_headers["Authorization"] = "Bearer secret-admin" + client.default_headers["Content-Type"] = "application/json" + + it "successfully executes an internal call that passes and triggers side-effects" do + # 1. Create a user via standard HTTP + res_user = client.post("/users", %({"name":"Alice"})) + user_id = res_user.json["id"].as_s + + # 2. Create a post. + # This will trigger `ValidateAuthorExists` (calls /users) + # and `CreateAuditLog` (calls /audit). + res_post = client.post("/posts", %({"author_id":"#{user_id}", "content":"Hello World"})) + res_post.status.should eq(201) + post_id = res_post.json["id"].as_s + + # 3. Verify the Audit service was successfully called internally + res_audit = client.get("/audit") + logs = res_audit.json.as_a + + logs.size.should eq(1) + logs.first["action"].as_s.should eq("Created Post") + logs.first["resource_id"].as_s.should eq(post_id) + end + + it "fails cleanly when an internal call raises an exception" do + # Try to create a post for a user that does not exist. + # The internal `ctx.call("/users", :get, id: "999")` will 404, throwing an Exception, + # which the rule rescues and converts into a 422. + res = client.post("/posts", %({"author_id":"999", "content":"Ghost Post"})) + + res.status.should eq(422) + res.json["details"]["author_id"].as_s.should eq("Author does not exist") + end + + it "propagates the context store to internal calls" do + # We create an isolated app for this test so we don't violate the `freeze_rules!` + # lock triggered by the HTTP requests in the previous tests. + isolated_app = Alumna::App.new + + captured_store_val = nil + + isolated_app.use "/audit", Alumna.memory(Alumna::Schema.new) { + before do |ctx| + captured_store_val = ctx.store["current_user_id"]? + nil + end + } + + isolated_app.before AuthenticateGlobal + + isolated_client = Alumna::Testing::AppClient.new(isolated_app) + isolated_client.default_headers["Authorization"] = "Bearer secret-admin" + + # Trigger a find on audit natively. Since the provider is "rest", + # the auth rule runs, sets the store, and the audit rule reads it. + isolated_client.get("/audit") + + captured_store_val.should eq("1") + end +end diff --git a/spec/service/context_spec.cr b/spec/service/context_spec.cr index 0b89147..d615bb8 100644 --- a/spec/service/context_spec.cr +++ b/spec/service/context_spec.cr @@ -113,4 +113,78 @@ describe Alumna::RuleContext do ctx.data_bool?("age").should be_nil end end + + # Add this inside the `describe Alumna::RuleContext do` block: + + describe "#call (Internal Routing)" do + it "dispatches internally to another service using symbols" do + app = Alumna::App.new + app.use "/target", Alumna::MemoryAdapter.new(Alumna::Schema.new.str("name")) + + # Start an initial context + ctx = Alumna::Testing.build_ctx(app: app) + + # Make the internal call + result = ctx.call("/target", :create, {"name" => "SubResource"} of String => Alumna::AnyData) + + result.should_not be_nil + result.as(Hash)["name"].should eq("SubResource") + result.as(Hash).has_key?("id").should be_true + end + + it "inherits the ctx.store so authentication passes down" do + app = Alumna::App.new + + # Target service expects "user" in the store + app.use "/secure", Alumna.memory(Alumna::Schema.new) { + before do |c| + c.store["user"]? ? nil : Alumna::ServiceError.unauthorized + end + } + + ctx = Alumna::Testing.build_ctx(app: app) + ctx.store["user"] = "Admin" # Authenticate the parent request + + # Internal call should succeed because the store is copied + result = ctx.call("/secure", :find) + result.as(Array).should be_empty # Memory adapter returns [] on empty find + end + + it "sets provider to 'internal' and http_method to 'INTERNAL'" do + app = Alumna::App.new + captured_provider = "" + + app.use "/probe", Alumna.memory(Alumna::Schema.new) { + before do |c| + captured_provider = c.provider + c.result = {"ok" => true} of String => Alumna::AnyData + nil + end + } + + ctx = Alumna::Testing.build_ctx(app: app) + ctx.call("/probe", :find) + + captured_provider.should eq("internal") + end + + it "raises ArgumentError if the internal path does not exist" do + ctx = Alumna::Testing.build_ctx + expect_raises(ArgumentError, /Internal service not found/) do + ctx.call("/nowhere", :find) + end + end + + it "raises an Exception if the internal service returns a ServiceError" do + app = Alumna::App.new + app.use "/fail", Alumna.memory(Alumna::Schema.new) { + before { |_c| Alumna::ServiceError.bad_request("Custom failure") } + } + + ctx = Alumna::Testing.build_ctx(app: app) + expect_raises(Exception, /Internal call to \/fail failed: 400 Custom failure/) do + ctx.call("/fail", :find) + end + end + end end diff --git a/src/alumna.cr b/src/alumna.cr index be0449d..b78cf20 100644 --- a/src/alumna.cr +++ b/src/alumna.cr @@ -55,6 +55,12 @@ require "./http/router" require "./http/responder" module Alumna + # Overload without a block + def self.memory(schema : Schema? = nil) + MemoryAdapter.new(schema) + end + + # Overload with a block def self.memory(schema : Schema? = nil, &) svc = MemoryAdapter.new(schema) # no block with svc yield # scope = svc diff --git a/src/app.cr b/src/app.cr index 74729df..6e934f1 100644 --- a/src/app.cr +++ b/src/app.cr @@ -120,9 +120,10 @@ module Alumna end def listen( - port : Int32 = 3000, + port : Int32? = 3000, *, host : String = "127.0.0.1", + unix_socket : String? = nil, trusted_proxies : TrustedProxies = nil, workers : Int32? = nil, shutdown_timeout : Time::Span? = 10.seconds, @@ -169,13 +170,24 @@ module Alumna end # reuse_port is no longer needed since Crystal threads share the same socket - server.bind_tcp(host, port) + server.bind_tcp(host, port) if port - display_host = host.includes?(':') ? "[#{host}]" : host - puts "Listening on http://#{display_host}:#{port}#{workers_msg}" + if unix_path = unix_socket + File.delete?(unix_path) # Clean up previous socket if left behind + server.bind_unix(unix_path) + end + + if port + display_host = host.includes?(':') ? "[#{host}]" : host + puts "Listening on http://#{display_host}:#{port}#{workers_msg}" - if host == "0.0.0.0" || host == "::" - STDERR.puts "Warning: binding to #{host} exposes the server on all interfaces" + if host == "0.0.0.0" || host == "::" + STDERR.puts "Warning: binding to #{host} exposes the server on all interfaces" + end + end + + if unix_socket + puts "Listening on unix://#{unix_socket}#{port ? "" : workers_msg}" end # LCOV_EXCL_START - kcov wrongly misses this block @@ -202,6 +214,11 @@ module Alumna puts "Graceful shutdown complete." end end + + # Cleanup unix socket to prevent dead files + if unix_path = unix_socket + File.delete?(unix_path) + end end end end diff --git a/src/http/router.cr b/src/http/router.cr index a9c389c..756e969 100644 --- a/src/http/router.cr +++ b/src/http/router.cr @@ -60,6 +60,7 @@ module Alumna phase: RulePhase::Before, http_method: request.method, remote_ip: remote_ip(http_ctx), + provider: resolve_provider(http_ctx), params: ParamsView.new(request.query_params), headers: HeadersView.new(request.headers), id: id, @@ -257,6 +258,10 @@ module Alumna ip = ctx.request.headers["X-Real-IP"]?.try(&.strip) ip if ip && Socket::IPAddress.valid?(ip) end + + private def resolve_provider(ctx : HTTP::Server::Context) : String + ctx.request.remote_address.is_a?(Socket::UNIXAddress) ? "local" : "rest" + end end end end diff --git a/src/service/context.cr b/src/service/context.cr index fb26204..eb8dc8a 100644 --- a/src/service/context.cr +++ b/src/service/context.cr @@ -18,10 +18,10 @@ module Alumna getter phase : RulePhase getter http_method : String getter remote_ip : String + getter provider : String + getter id : String? property params : Http::ParamsView - property provider : String - property id : String? property data : Hash(String, AnyData) property result : ServiceResult = nil property error : ServiceError? = nil @@ -63,6 +63,51 @@ module Alumna @result_set = true end + # Dispatches a request to another internal service, bypassing the HTTP network stack + # but still running through the target service's schema validations and rules. + def call( + path : String, + method : ServiceMethod | Symbol, + data : Hash(String, AnyData) = {} of String => AnyData, + id : String? = nil, + ) : ServiceResult + target_service = app.services[path]? + raise ArgumentError.new("Internal service not found at path: #{path}") unless target_service + + parsed_method = method.is_a?(ServiceMethod) ? method : ServiceMethod.parse(method.to_s.capitalize) + + internal_ctx = RuleContext.new( + app: app, + service: target_service, + path: path, + method: parsed_method, + phase: RulePhase::Before, + http_method: "INTERNAL", + remote_ip: remote_ip, + provider: "internal", + params: params, # Inherit params view + headers: headers, # Inherit headers view + id: id, + data: data + ) + + # Shallow copy the store so authenticated users/transaction IDs flow down, + # but downstream mutations don't pollute the parent request state. + if s = @store + s.each { |k, v| internal_ctx.store[k] = v } + end + + app.dispatch(target_service, internal_ctx) + + # If the sub-service threw an error (e.g., validation failed), re-raise it + # so the calling rule/service can rescue it, or let it halt the current pipeline. + if err = internal_ctx.error + raise Exception.new("Internal call to #{path} failed: #{err.status} #{err.message}") + end + + internal_ctx.result + end + @[AlwaysInline] def result_set? : Bool @result_set