使用 Pulumi、SSE 与 Headless UI 构建 iOS 实时开发者状态面板的实践

我们的 iOS 开发团队长期面临一个信息孤岛问题：CI/CD 的构建状态、功能开关（Feature Flag）的实时配置、测试环境的健康度监控，这些关键信息散落在 Jenkins、LaunchDarkly 和 Grafana 等多个平台。工程师需要频繁在这些 Web 面板间切换，不仅效率低下，而且在移动端体验极差。更重要的是，关键状态的变更（如构建失败或功能开关线上关闭）无法被实时、主动地推送给相关人员。我们需要一个统一的、原生的、实时的状态中心。

最初的构想是一个简单的 iOS App，通过轮询各个服务的 API 来获取数据。但这很快被否决了。轮询不仅浪费设备电量和网络资源，而且信息延迟严重。更棘手的是 UI 的问题，为每一种数据源（构建、功能开关、服务器状态）都定制一套 UI 视图，意味着大量的重复工作和极差的可扩展性。每当需要监控一个新的指标，UI 和业务逻辑都得重写。

这个困境迫使我们重新思考整个架构。我们需要的是一个系统，它应该具备以下特质：

1. 实时性: 状态变更必须能从服务端实时推送到客户端，而非客户端主动拉取。
2. UI 解耦: UI 组件应该与具体的业务数据模型解耦，成为可复用的“渲染器”。
3. 基础设施即代码 (IaC): 后端服务必须能够被自动化、可复现地部署和管理，避免手动配置带来的风险。

基于这些考量，我们最终确定了一套技术栈：Server-Sent Events (SSE) 用于实现轻量级的服务端推送，Headless UI 模式在 iOS 端（使用 SwiftUI）实现视图与逻辑的分离，以及 Pulumi 利用我们团队熟悉的 TypeScript 来完成整个云端基础设施的声明式管理。

第一步：用 Pulumi 定义可流式响应的后端服务

我们的核心需求是服务端推送，WebSockets 似乎是首选，但它是一个相对复杂的协议，需要处理连接生命周期、心跳等。在真实项目中，对于我们这种单向（Server -> Client）通信的场景，SSE 是一个更简单、更稳健的选择。它基于标准的 HTTP，客户端实现非常简单，并且内置了断线重连机制。

我们决定使用 AWS Lambda 和 API Gateway V2 来构建这个 SSE 服务。Lambda 支持流式响应（Response Streaming），这正是 SSE 所需要的——一个永不关闭的 HTTP 连接，服务器可以随时向其写入数据。手动在 AWS控制台配置这一切是繁琐且不可靠的，因此我们选择 Pulumi。

以下是使用 TypeScript 定义整个后端基础设施的 Pulumi 代码。这里的关键在于配置 API Gateway 的集成，使其指向一个支持流式调用的 Lambda 函数。

// pulumi/index.ts
import * as aws from "@pulumi/aws";
import * as pulumi from "@pulumi/pulumi";
import * as path from "path";

// 1. 定义 IAM Role，允许 Lambda 执行和写入 CloudWatch Logs
const lambdaRole = new aws.iam.Role("sseLambdaRole", {
    assumeRolePolicy: aws.iam.assumeRolePolicyForPrincipal({
        Service: "lambda.amazonaws.com",
    }),
});

new aws.iam.RolePolicyAttachment("lambdaLogs", {
    role: lambdaRole.name,
    policyArn: aws.iam.ManagedPolicy.AWSLambdaBasicExecutionRole,
});

// 2. 定义 Lambda 函数，关键点是 `invokeMode: "RESPONSE_STREAM"`
// 这告诉 Lambda 我们需要一个流式响应的调用模式
const sseLambda = new aws.lambda.Function("sseStreamHandler", {
    runtime: aws.lambda.Runtime.NodeJS18dX,
    role: lambdaRole.arn,
    handler: "index.handler",
    code: new pulumi.asset.AssetArchive({
        ".": new pulumi.asset.FileArchive(path.join(__dirname, "../lambda")),
    }),
    timeout: 300, // SSE 连接需要较长的超时时间
    memorySize: 256,
    invokeMode: "RESPONSE_STREAM", // 核心配置：启用流式响应
});

// 3. 创建 API Gateway (HTTP API - V2)，它比 REST API 更轻量、成本更低
const api = new aws.apigatewayv2.Api("sseApi", {
    protocolType: "HTTP",
    name: "RealtimeStatusDashboardAPI",
});

// 4. 配置 API Gateway 与 Lambda 的集成
const integration = new aws.apigatewayv2.Integration("sseIntegration", {
    apiId: api.id,
    integrationType: "AWS_PROXY",
    integrationUri: sseLambda.invokeArn,
    payloadFormatVersion: "2.0",
});

// 5. 定义路由，将 GET /stream 请求转发到 Lambda 集成
new aws.apigatewayv2.Route("sseRoute", {
    apiId: api.id,
    routeKey: "GET /stream",
    target: pulumi.interpolate`integrations/${integration.id}`,
});

// 6. 创建部署阶段，使 API 可被公开访问
const stage = new aws.apigatewayv2.Stage("sseStage", {
    apiId: api.id,
    name: "$default",
    autoDeploy: true,
});

// 7. 授予 API Gateway 调用 Lambda 的权限
new aws.lambda.Permission("apiGatewayInvokePermission", {
    action: "lambda:InvokeFunction",
    function: sseLambda.name,
    principal: "apigateway.amazonaws.com",
    sourceArn: pulumi.interpolate`${api.executionArn}/*/*`,
});

// 导出可供客户端访问的 URL
export const apiUrl = stage.invokeUrl;
export const streamEndpoint = pulumi.interpolate`${stage.invokeUrl}/stream`;

这段 Pulumi 代码做了几件关键的事情：

• 明确将 Lambda 的 invokeMode 设置为 RESPONSE_STREAM，这是实现 SSE 的基础。
• 使用 API Gateway V2 (HTTP API)，因为它对流式负载有更好的支持且成本更低。
• 完整定义了从路由、集成到权限的所有资源，一个 pulumi up 命令就能部署整套服务。

接下来是 Lambda 函数本身的代码，它负责生成 SSE 事件流。一个常见的错误是直接返回一个完整的 HTTP 响应，但在流式模式下，我们需要操作一个可写的流对象。

// lambda/index.ts
import { Writable } from "stream";

// AWS 为流式响应提供了一个全局的 'awslambda' 对象
declare const awslambda: {
    streamifyResponse: (
        handler: (
            event: any,
            responseStream: Writable,
            context: any
        ) => Promise<void>
    ) => any;
};

// SSE 事件的数据结构
interface SSEEvent {
    event: string;
    data: object;
    id?: string;
}

const writeSseEvent = (stream: Writable, event: SSEEvent) => {
    // 生产环境中，日志记录是必须的
    console.log(`Sending event: ${event.event}`, JSON.stringify(event.data));

    let message = "";
    if (event.id) {
        message += `id: ${event.id}\n`;
    }
    if (event.event) {
        message += `event: ${event.event}\n`;
    }
    // data 字段必须是 JSON 字符串
    message += `data: ${JSON.stringify(event.data)}\n\n`; // 注意末尾的双换行符

    stream.write(message);
};

// 模拟外部事件源，例如来自 CI 系统或数据库变更流
const eventGenerator = (stream: Writable) => {
    // 模拟 CI 构建状态更新
    setInterval(() => {
        const buildId = `build-${Math.floor(Math.random() * 1000)}`;
        const statuses = ["RUNNING", "SUCCESS", "FAILED"];
        const randomStatus = statuses[Math.floor(Math.random() * statuses.length)];
        
        writeSseEvent(stream, {
            event: "ci_build_status",
            data: { 
                id: buildId,
                name: "iOS-App-Release-Pipeline", 
                status: randomStatus,
                timestamp: new Date().toISOString() 
            },
            id: new Date().getTime().toString(),
        });
    }, 5000);

    // 模拟功能开关状态更新
    setInterval(() => {
        const flagName = "new-checkout-flow";
        const isEnabled = Math.random() > 0.5;

        writeSseEvent(stream, {
            event: "feature_flag_update",
            data: { 
                name: flagName, 
                enabled: isEnabled,
                environment: "production",
                updatedBy: "auto-system"
            },
            id: new Date().getTime().toString(),
        });
    }, 8000);
};

export const handler = awslambda.streamifyResponse(
    async (event, responseStream, context) => {
        // 设置 SSE 所需的 HTTP Headers
        responseStream.write(
            "HTTP/1.1 200 OK\r\n" +
            "Content-Type: text/event-stream\r\n" +
            "Cache-Control: no-cache\r\n" +
            "Connection: keep-alive\r\n\r\n"
        );

        // 发送一个 keep-alive 消息，防止客户端或中间代理因超时而断开连接
        const keepAliveInterval = setInterval(() => {
            // SSE 规范中，一个冒号开头的行是注释，通常用作 ping
            responseStream.write(":keep-alive\n\n");
        }, 20000);

        // 监听连接关闭事件，清理定时器
        responseStream.on('close', () => {
            console.log("Stream closed. Cleaning up resources.");
            clearInterval(keepAliveInterval);
            // 在真实应用中，这里也应停止 eventGenerator
        });

        // 启动事件生成器
        eventGenerator(responseStream);

        // 在 Lambda 超时之前，我们需要一个机制来保持函数运行
        // 在真实项目中，这里会连接到一个消息队列（如 SQS/SNS）或数据库流
        await new Promise(resolve => setTimeout(resolve, context.getRemainingTimeInMillis() - 500));
        
        // 正常情况下，这个函数不会执行到这里，除非即将超时
        responseStream.end();
    }
);

这里的坑在于：awslambda.streamifyResponse 是 AWS 提供的一个包装器，它改变了 handler 的签名。我们得到的不再是 callback，而是一个可写的 responseStream 对象。我们必须手动写入 HTTP 头，并确保 Content-Type 是 text/event-stream。此外，定期的 keep-alive 注释对于维持长连接在生产环境中至关重要。

第二步：在 SwiftUI 中设计可复用的 Headless UI 组件

现在后端已经就绪，我们转向客户端。传统的 MVC/MVVM 模式下，我们可能会创建一个 CIBuildView 和一个 CIBuildViewModel。这种紧耦合的方式导致了我们最初遇到的可扩展性问题。

Headless UI 的核心思想是，将组件的逻辑和状态管理（“大脑”）与其视觉表现（“皮肤”）彻底分离。在 SwiftUI 中，我们可以通过协议和泛型来实现这一点。

首先，我们定义一个通用的 StatusItem 协议，它抽象了所有状态项共有的特征。

// Models/StatusItem.swift
import Foundation
import SwiftUI

// 定义状态的视觉表现
enum StatusColor {
    case green, yellow, red, gray
    
    var color: Color {
        switch self {
        case .green: return .green
        case .yellow: return .orange
        case .red: return .red
        case .gray: return .gray
        }
    }
}

// 抽象的状态项协议
// 任何希望被 StatusCardView 渲染的数据模型都需要遵守此协议
protocol StatusItem: Identifiable, Hashable {
    associatedtype Detail: Hashable
    
    var id: String { get }
    var title: String { get }
    var status: StatusColor { get }
    var details: [Detail] { get }
    
    // 用于将具体模型转换为通用渲染所需模型的工厂方法
    static func from(dto: Decodable) -> Self?
}

然后，我们可以为具体的数据源（如 CI 构建和功能开关）创建符合此协议的模型。

// Models/CIBuild.swift
struct CIBuild: Decodable, StatusItem {
    let buildId: String
    let pipelineName: String
    let buildStatus: String
    let timestamp: String

    // MARK: - StatusItem Conformance
    var id: String { buildId }
    var title: String { pipelineName }
    
    var status: StatusColor {
        switch buildStatus {
        case "SUCCESS": return .green
        case "RUNNING": return .yellow
        case "FAILED": return .red
        default: return .gray
        }
    }
    
    // 这里的 Detail 是一个简单的 KeyValue 对
    struct Detail: Hashable {
        let key: String
        let value: String
    }
    
    var details: [Detail] {
        [
            .init(key: "Status", value: buildStatus),
            .init(key: "Time", value: timestamp)
        ]
    }
    
    static func from(dto: Decodable) -> CIBuild? {
        // 在真实项目中，这里会进行安全的类型转换和数据校验
        return dto as? CIBuildDTO
    }
}

// DTO (Data Transfer Object) 用于解码 SSE 发来的 JSON
struct CIBuildDTO: Decodable {
    let id: String
    let name: String
    let status: String
    let timestamp: String
}

最关键的部分是通用的 “Headless” 视图。StatusCardView 并不知道它正在渲染的是一个 CI 构建还是别的什么东西。它只关心传入的模型是否遵守 StatusItem 协议。

// Views/StatusCardView.swift
import SwiftUI

// 这是一个完全通用的视图，它不依赖任何具体的模型类型
// 它只要求传入的 `item` 遵守 `StatusItem` 协议
struct StatusCardView<Item: StatusItem>: View where Item.Detail == CIBuild.Detail {
    let item: Item
    
    var body: some View {
        HStack(alignment: .top, spacing: 16) {
            // 左侧的状态指示器
            Rectangle()
                .fill(item.status.color)
                .frame(width: 8)

            VStack(alignment: .leading, spacing: 8) {
                Text(item.title)
                    .font(.headline)
                    .fontWeight(.bold)
                
                // 渲染详情
                ForEach(item.details, id: \.self) { detail in
                    HStack {
                        Text("\(detail.key):")
                            .font(.caption)
                            .foregroundColor(.secondary)
                        Text(detail.value)
                            .font(.caption.monospaced())
                    }
                }
            }
            .padding(.vertical, 8)
            
            Spacer()
        }
        .background(Color(.secondarySystemBackground))
        .cornerRadius(12)
        .frame(maxWidth: .infinity)
    }
}

这种架构的威力在于，当我们想添加一个新的状态源（例如，服务器 CPU 负载监控），我们只需要：

1. 创建一个新的 ServerHealth 模型，并让它遵守 StatusItem 协议。
2. 在 SSE 后端添加一个新的事件类型 server_health_update。
3. 在客户端的数据处理层添加对这个新事件的解码逻辑。

UI 代码一行都不用改。StatusCardView 可以直接渲染 ServerHealth 实例。

第三步：实现一个稳健的 Swift SSE 客户端

iOS 原生的 URLSession 对 SSE 提供了良好的支持。我们需要创建一个专门的客户端类来管理连接、解析事件和处理重连。一个常见的错误是简单地使用 URLSession.dataTask 并期望一切顺利。在生产环境中，网络是不稳定的，必须实现可靠的错误处理和重连逻辑。

// Services/SSEClient.swift
import Foundation
import Combine

enum SSEClientError: Error {
    case invalidURL
    case networkError(Error)
    case parsingError
}

struct SSEMessage {
    let id: String?
    let event: String
    let data: String
}

class SSEClient: NSObject, URLSessionDataDelegate {
    private var urlSession: URLSession!
    private var dataTask: URLSessionDataTask?
    private let url: URL
    
    private let messageSubject = PassthroughSubject<SSEMessage, SSEClientError>()
    var publisher: AnyPublisher<SSEMessage, SSEClientError> {
        messageSubject.eraseToAnyPublisher()
    }
    
    // 用于重连的参数
    private var retryCount = 0
    private let maxRetries = 5
    private let baseRetryDelay: TimeInterval = 1.0

    init(url: URL) {
        self.url = url
        super.init()
        // 使用自定义 delegate 队列，避免阻塞主线程
        let configuration = URLSessionConfiguration.default
        configuration.timeoutIntervalForRequest = TimeInterval(300)
        self.urlSession = URLSession(configuration: configuration, delegate: self, delegateQueue: OperationQueue())
    }
    
    func connect() {
        guard dataTask == nil else {
            print("Connection already active.")
            return
        }
        print("Connecting to SSE stream...")
        var request = URLRequest(url: url)
        request.httpMethod = "GET"
        request.setValue("text/event-stream", forHTTPHeaderField: "Accept")
        request.setValue("no-cache", forHTTPHeaderField: "Cache-Control")
        
        dataTask = urlSession.dataTask(with: request)
        dataTask?.resume()
    }
    
    func disconnect() {
        dataTask?.cancel()
        dataTask = nil
        print("Disconnected from SSE stream.")
    }
    
    private func handleReconnect() {
        dataTask = nil
        guard retryCount < maxRetries else {
            print("Max retries reached. Giving up.")
            messageSubject.send(completion: .failure(.networkError(URLError(.cancelled)))) // Or a custom error
            return
        }
        
        // 指数退避重连策略
        let delay = baseRetryDelay * pow(2.0, Double(retryCount))
        retryCount += 1
        
        print("Connection lost. Reconnecting in \(delay) seconds... (Attempt \(retryCount))")
        
        DispatchQueue.global().asyncAfter(deadline: .now() + delay) { [weak self] in
            self?.connect()
        }
    }
    
    // MARK: - URLSessionDataDelegate
    
    func urlSession(_ session: URLSession, dataTask: URLSessionDataTask, didReceive data: Data) {
        guard let messageString = String(data: data, encoding: .utf8) else {
            return
        }
        
        // 一个数据包可能包含多个 SSE 消息，必须按 "\n\n" 分割
        let lines = messageString.components(separatedBy: "\n\n")
        
        for line in lines where !line.isEmpty {
            parseAndPublish(message: line)
        }
    }

    private func parseAndPublish(message: String) {
        var event = "message" // 默认事件类型
        var data = ""
        var id: String?

        let fields = message.components(separatedBy: "\n")
        for field in fields {
            if field.hasPrefix("event:") {
                event = String(field.dropFirst(6)).trimmingCharacters(in: .whitespaces)
            } else if field.hasPrefix("data:") {
                data = String(field.dropFirst(5)).trimmingCharacters(in: .whitespaces)
            } else if field.hasPrefix("id:") {
                id = String(field.dropFirst(3)).trimmingCharacters(in: .whitespaces)
            }
        }
        
        guard !data.isEmpty else {
            // 可能是 keep-alive 注释，忽略
            return
        }
        
        let sseMessage = SSEMessage(id: id, event: event, data: data)
        // 回到主线程发布事件
        DispatchQueue.main.async {
            self.messageSubject.send(sseMessage)
        }
    }
    
    func urlSession(_ session: URLSession, task: URLSessionTask, didCompleteWithError error: Error?) {
        if let error = error {
            print("SSE Client Error: \(error.localizedDescription)")
            // 如果不是用户主动取消，则尝试重连
            if (error as? URLError)?.code != .cancelled {
                handleReconnect()
            }
        } else {
             // 正常情况下，长连接不应该在这里完成，如果完成了，也视为异常断开
             print("Stream completed unexpectedly. Attempting to reconnect.")
             handleReconnect()
        }
    }
    
    func urlSession(_ session: URLSession, dataTask: URLSessionDataTask, didReceive response: URLResponse, completionHandler: @escaping (URLSession.ResponseDisposition) -> Void) {
        // 连接成功，重置重连计数器
        print("SSE stream connected successfully.")
        self.retryCount = 0
        completionHandler(.allow)
    }
}

这个 SSEClient 的实现考虑了几个生产级别的要点：

1. 指数退避重连: handleReconnect 方法实现了指数退避算法，避免了在服务器故障时发起“拒绝服务”式的密集重连。
2. 委托处理: 通过 URLSessionDataDelegate，我们可以更精细地处理数据流和连接状态，例如在连接成功时重置重连计数器。
3. 消息解析: 正确处理了可能在一个数据包中出现的多个 SSE 消息，并解析 event, data, id 字段。
4. 线程安全: 使用 Combine 的 PassthroughSubject 并在主线程上发布事件，确保 UI 更新是安全的。

整合：从数据流到动态 UI

最后一步是将所有部分连接起来。我们需要一个主视图模型 (DashboardViewModel) 来管理 SSEClient，监听事件，解码数据，并更新驱动 UI 的状态。

sequenceDiagram
    participant CI/CD Server
    participant SSE Lambda
    participant iOS App (SSEClient)
    participant DashboardViewModel
    participant SwiftUI View
    
    CI/CD Server->>SSE Lambda: (via Webhook/Event) Trigger Event: Build Failed
    SSE Lambda-->>iOS App (SSEClient): SSE Event (event: ci_build_status, data: {...})
    iOS App (SSEClient)-->>DashboardViewModel: Publishes parsed SSEMessage
    DashboardViewModel->>DashboardViewModel: Decodes data into CIBuild model
    DashboardViewModel->>DashboardViewModel: Updates @Published items array
    SwiftUI View-->>DashboardViewModel: Reacts to @Published change
    SwiftUI View->>SwiftUI View: Re-renders body with updated items

这是 DashboardViewModel 的实现：

// ViewModels/DashboardViewModel.swift
import Foundation
import Combine

class DashboardViewModel: ObservableObject {
    // 使用 AnyHashable 包装不同类型的 StatusItem
    @Published var statusItems: [AnyHashable] = []
    
    private var sseClient: SSEClient?
    private var cancellables = Set<AnyCancellable>()
    
    private let decoder = JSONDecoder()

    func connect(endpoint: String) {
        guard let url = URL(string: endpoint) else {
            print("Invalid SSE endpoint URL")
            return
        }
        
        sseClient = SSEClient(url: url)
        sseClient?.publisher
            .receive(on: DispatchQueue.main)
            .sink(receiveCompletion: { completion in
                if case .failure(let error) = completion {
                    print("SSE stream failed: \(error)")
                }
            }, receiveValue: { [weak self] message in
                self?.handleSseMessage(message)
            })
            .store(in: &cancellables)
        
        sseClient?.connect()
    }
    
    private func handleSseMessage(_ message: SSEMessage) {
        guard let data = message.data.data(using: .utf8) else { return }

        do {
            var newItem: AnyHashable?
            
            // 根据事件类型解码为不同的数据模型
            switch message.event {
            case "ci_build_status":
                let dto = try decoder.decode(CIBuildDTO.self, from: data)
                let build = CIBuild(buildId: dto.id, pipelineName: dto.name, buildStatus: dto.status, timestamp: dto.timestamp)
                newItem = build
                
            case "feature_flag_update":
                // 类似地解码 FeatureFlagDTO 并创建 FeatureFlag 模型...
                break
                
            default:
                print("Received unknown event type: \(message.event)")
            }
            
            if let item = newItem {
                updateOrAppendItem(item)
            }
            
        } catch {
            print("Failed to decode SSE data: \(error)")
        }
    }
    
    private func updateOrAppendItem(_ item: AnyHashable) {
        // 为了简化，这里假设 item 遵守 Identifiable。
        // 一个更健壮的实现会检查 AnyHashable 的底层类型。
        guard let identifiableItem = item as? (any Identifiable) else { return }

        if let index = statusItems.firstIndex(where: { ($0 as? (any Identifiable))?.id == identifiableItem.id }) {
            statusItems[index] = item
        } else {
            statusItems.insert(item, at: 0)
        }
    }

    deinit {
        sseClient?.disconnect()
    }
}

最终，SwiftUI 视图变得异常简洁：

// Views/DashboardView.swift
struct DashboardView: View {
    @StateObject private var viewModel = DashboardViewModel()
    
    // 这个 URL 来自 Pulumi 的输出
    private let sseEndpoint = "YOUR_PULUMI_OUTPUT_STREAM_ENDPOINT"

    var body: some View {
        NavigationView {
            ScrollView {
                LazyVStack(spacing: 16) {
                    ForEach(viewModel.statusItems, id: \.self) { item in
                        // 根据 item 的具体类型，动态选择渲染视图
                        // 这里展示了 Headless UI 架构的灵活性
                        if let build = item as? CIBuild {
                            StatusCardView(item: build)
                        } else if let flag = item as? FeatureFlag {
                            // StatusCardView(item: flag)
                            // 假设 FeatureFlag 也遵守 StatusItem
                        }
                    }
                }
                .padding()
            }
            .navigationTitle("Developer Dashboard")
            .onAppear {
                viewModel.connect(endpoint: sseEndpoint)
            }
        }
    }
}

这个方案的最终成果是一个可维护、可扩展的实时监控系统。iOS 客户端的 UI 组件是“无脑的”，它们只负责渲染。所有的数据转换、业务逻辑都封装在模型和视图模型中。而后端，由于 Pulumi 的存在，任何工程师都可以检出代码，用一行命令在自己的 AWS 账户中部署一套完全相同的、隔离的测试环境。

当前方案的局限性与未来展望

尽管这套架构解决了我们最初的核心问题，但它并非完美无缺。
首先，SSE 是单向通信。如果需要客户端发起操作（例如，触发一次 CI 重建），仍然需要一个独立的 REST 或 GraphQL API 来处理这些命令。这增加了架构的复杂性。

其次，我们使用的 AWS Lambda + API Gateway 方案虽然简单且无服务器化，但在超大规模连接数（数万个并发客户端）的场景下，成本和性能可能不是最优的。届时可能需要评估更专业的服务，如 AWS IoT Core for MQTT 或自建基于 K8s 的推送集群。

在客户端，目前的 DashboardViewModel 中的类型判断 if let build = item as? CIBuild 还是有些僵硬。未来的迭代可以引入一个注册表模式，根据 event 类型动态查找对应的解码器和模型工厂，实现一个完全由数据驱动的 UI 渲染管线。

最后，基础设施代码虽然大大提升了可靠性，但也引入了对 Pulumi 状态文件（state file）的管理需求。在团队协作中，必须使用 S3 等后端来存储和锁定状态文件，避免并发修改基础设施导致的状态冲突。