RPC框架：服务间通信与接口定义

// project_service.proto
syntax = "proto3";

package aiide.project.v1;

option go_package = "github.com/aiide/backend/gen/go/project/v1";
option java_multiple_files = true;
option java_package = "com.aiide.project.v1";

// 项目状态枚举
enum ProjectStatus {
  PROJECT_STATUS_UNSPECIFIED = 0;
  PROJECT_STATUS_INITIALIZING = 1;
  PROJECT_STATUS_ACTIVE = 2;
  PROJECT_STATUS_SUSPENDED = 3;
  PROJECT_STATUS_DELETED = 4;
}

// 项目信息
message Project {
  string id = 1;
  string name = 2;
  string owner_id = 3;
  repeated string collaborator_ids = 4;
  ProjectStatus status = 5;
  int64 created_at = 6;
  int64 updated_at = 7;
  map<string, string> metadata = 8;
}

// 创建项目请求
message CreateProjectRequest {
  string name = 1;
  string description = 2;
  repeated string collaborator_ids = 3;
  map<string, string> metadata = 4;
}

// 创建项目响应
message CreateProjectResponse {
  Project project = 1;
}

// 获取项目请求
message GetProjectRequest {
  string id = 1;
}

// 获取项目响应
message GetProjectResponse {
  Project project = 1;
}

// 列出用户项目请求
message ListUserProjectsRequest {
  string user_id = 1;
  int32 page_size = 2;
  string page_token = 3;
}

// 列出用户项目响应
message ListUserProjectsResponse {
  repeated Project projects = 1;
  string next_page_token = 2;
  int32 total_count = 3;
}

// 项目服务定义
service ProjectService {
  // 创建项目
  rpc CreateProject(CreateProjectRequest) returns (CreateProjectResponse);
  
  // 获取项目
  rpc GetProject(GetProjectRequest) returns (GetProjectResponse);
  
  // 更新项目
  rpc UpdateProject(UpdateProjectRequest) returns (UpdateProjectResponse);
  
  // 删除项目
  rpc DeleteProject(DeleteProjectRequest) returns (DeleteProjectResponse);
  
  // 列出用户项目（支持分页）
  rpc ListUserProjects(ListUserProjectsRequest) returns (ListUserProjectsResponse);
  
  // 流式获取项目变化（双向流）
  rpc WatchProjects(WatchProjectsRequest) returns (stream ProjectChangeEvent);
}

// 更新项目请求
message UpdateProjectRequest {
  string id = 1;
  string name = 2;
  ProjectStatus status = 3;
  map<string, string> metadata = 4;
}

// 更新项目响应
message UpdateProjectResponse {
  Project project = 1;
}

// 删除项目请求
message DeleteProjectRequest {
  string id = 1;
}

// 删除项目响应
message DeleteProjectResponse {
  bool success = 1;
}

// 监听项目变更请求
message WatchProjectsRequest {
  repeated string project_ids = 1;
}

// 项目变更事件
message ProjectChangeEvent {
  string project_id = 1;
  ChangeType change_type = 2;
  Project project = 3;
  int64 timestamp = 4;
  
  enum ChangeType {
    CHANGE_TYPE_UNSPECIFIED = 0;
    CHANGE_TYPE_CREATED = 1;
    CHANGE_TYPE_UPDATED = 2;
    CHANGE_TYPE_DELETED = 3;
  }
}

// 错误：永远不要重新使用已删除字段的编号
message Corrected {
  string name = 1;      // 原来是 id = 1, 已删除，但不能用 1
  int64 deleted_at = 2; // 新增字段
}

// 正确：保留已删除字段的编号为注释，永不重用
message Correct {
  // 旧版本: string id = 1; 已删除，不再使用
  string name = 2;      // 复用新的编号
  int64 deleted_at = 3;
}

// aiide_execution.thrift
namespace go aiide.execution.v1
namespace java com.aiide.execution.v1

// 执行状态
enum ExecutionStatus {
    PENDING = 0,
    RUNNING = 1,
    COMPLETED = 2,
    FAILED = 3,
    CANCELLED = 4,
    TIMEOUT = 5
}

// 代码执行请求
struct ExecutionRequest {
    1: required string execution_id,
    2: required string language,
    3: required string code,
    4: optional i32 timeout_ms = 30000,
    5: optional i64 memory_limit_mb = 256,
    6: optional map<string, string> environment,
    7: optional string stdin_input
}

// 代码执行响应
struct ExecutionResponse {
    1: required string execution_id,
    2: required ExecutionStatus status,
    3: optional string stdout,
    4: optional string stderr,
    5: optional i32 exit_code,
    6: optional i64 execution_time_ms,
    7: optional i64 memory_usage_mb,
    8: optional ExecutionError error
}

// 执行错误详情
struct ExecutionError {
    1: required i32 error_code,
    2: required string error_message,
    3: optional string stack_trace,
    4: optional map<string, string> metadata
}

// 执行服务
service ExecutionService {
    // 同步执行（简单RPC）
    ExecutionResponse execute(1: ExecutionRequest request) throws (1: ExecutionException ex),
    
    // 流式执行（服务端流）
    stream<ExecutionResponse> executeStream(1: ExecutionRequest request) throws (1: ExecutionException ex),
    
    // 批量执行
    list<ExecutionResponse> executeBatch(1: list<ExecutionRequest> requests) throws (1: ExecutionException ex),
    
    // 取消执行
    oneway void cancelExecution(1: string execution_id)
}

// 自定义异常
exception ExecutionException {
    1: i32 error_code,
    2: string error_message,
    3: string details
}

// 过度嵌套 - 导致Protobuf序列化层数过多
message ProjectWithDetails {
  Project project = 1;
  Owner owner = 2;
  repeated Collaborator collaborators = 3;
  Repository repository = 4;
  Settings settings = 5;
}

// 适度嵌套 - 根据业务边界划分
message Project {
  string id = 1;
  string name = 2;
  ProjectOwner owner = 3;
  ProjectSettings settings = 4;
}

// 推荐使用Oneof处理互斥字段
message ExecutionResponse {
  string execution_id = 1;
  oneof result {
    ExecutionSuccess success = 2;
    ExecutionFailure failure = 3;
  }
}

message ExecutionSuccess {
  string output = 1;
  int32 exit_code = 2;
}

message ExecutionFailure {
  string error_code = 1;
  string error_message = 2;
}

// 1. 字段编号一旦使用，绝对不能更改或删除
// 2. 添加新字段时，使用新的编号
message ApiResponse {
  string request_id = 1;      // 原始字段
  bool success = 2;          // 原始字段
  string message = 3;         // 原始字段
  
  // ========== 2024-01 新增字段 ==========
  map<string, string> headers = 100;   // 新增：响应头
  int64 server_timestamp = 101;       // 新增：服务器时间戳
  // ====================================
  
  // ========== 2024-06 新增字段 ==========
  RetryInfo retry_info = 200;         // 新增：重试信息
  // ====================================
}

// 3. 永远使用Optional而不是Required
// Required在Proto3中已被移除，但提醒我们：不能假设字段一定存在

// 4. 枚举值永远不要删除，只标记为 DEPRECATED
enum ProjectStatus {
  PROJECT_STATUS_UNSPECIFIED = 0;
  PROJECT_STATUS_INITIALIZING = 1;  // 已迁移到新的状态机
  reserved 1;  // 保留编号，防止误用
  
  PROJECT_STATUS_ACTIVE = 2;
  PROJECT_STATUS_SUSPENDED = 3;
  PROJECT_STATUS_ARCHIVED = 4;  // 新增
}

// 推荐的组织结构: {org}.{service}.{version}
// 这样可以保证跨组织的唯一性，避免命名冲突

package aiide.project.v1;        // 正确
package com.aiide.project.v1;    // 可接受

// 消息命名: PascalCase
message CreateProjectRequest {}  // 正确
message create_project_request {} // 错误

// 服务命名: PascalCase + Service后缀
service ProjectService {}        // 正确
service Project {}               // 错误
service projectService {}        // 错误

// 方法命名: PascalCase
rpc CreateProject(...) returns (...);  // 正确
rpc create_project(...) returns (...); // 错误

# 测试环境
CPU: Intel Xeon Gold 6248R @ 3.00GHz
Memory: 256GB DDR4
OS: Ubuntu 22.04 LTS
Go: 1.22.0
Java: OpenJDK 21

# 测试数据集
- Small Message: ~100 bytes (用户信息)
- Medium Message: ~1KB (项目信息)
- Large Message: ~100KB (代码片段 + AST)

// Varint编码原理：
// 数值越小，字节数越少
// 1字节：0-127 (无高位标记)
// 2字节：128-16383 (最高位标记为1表示后续还有字节)
//
// 数字 300 的编码过程：
// 300 = 10101100 (binary)
// 分成 7-bit 块: [1010110] [10]
// 添加延续位:   [1010110 1] [10 0] (最后一块延续位为0)
// 反转每块:     [1010110 1] [0000010]
// 结果: 0xAC 0x02 (hex)
//
// 对比：JSON直接编码为 "300" (3字节)，ProtoBuf只需2字节

package main

import (
    "encoding/binary"
    "fmt"
)

func encodeVarint(val uint64) []byte {
    var buf [10]byte
    var i int
    for i = 0; i < len(buf); i++ {
        if val >= 0x80 {
            buf[i] = byte(val) | 0x80 // 最高位1表示后续还有字节
            val >>= 7
        } else {
            buf[i] = byte(val)
            break
        }
    }
    return buf[:i+1]
}

func main() {
    // 编码测试
    tests := []uint64{0, 127, 128, 300, 16383, 16384, 1<<63 - 1}
    for _, v := range tests {
        encoded := encodeVarint(v)
        fmt.Printf("Value: %d -> [%v] (%d bytes)\n", v, encoded, len(encoded))
    }
}

// ProtoBuf二进制格式结构：
// [Field Tag] [Length] [Value]  或  [Field Tag] [Value]
//
// Field Tag = (field_number << 3) | wire_type
// Wire Types:
//   0 - Varint
//   1 - 64-bit
//   2 - Length-delimited (string, bytes, 嵌套消息)
//   5 - 32-bit

// 示例：消息 "name: "Alice", age: 30"
// 
// Field 1 (name, string): tag = (1 << 3) | 2 = 0x0A (10)
//   length = 5 (5字节)
//   value = "Alice"
// 
// Field 2 (age, int32): tag = (2 << 3) | 0 = 0x10 (16)
//   value = 30 -> 0x1E
//
// 二进制: 0x0A 0x05 0x41 0x6C 0x69 0x63 0x65 0x10 0x1E
// 字符串表示: "\n\x05Alice\x10\x1e"

package main

import (
    "encoding/json"
    "fmt"
    "time"
    
    "github.com/goccy/go-json"  // 第三方案例
    "github.com/json-iterator/go"  // json-iterator
)

type Project struct {
    ID        string            `json:"id"`
    Name      string            `json:"name"`
    OwnerID   string            `json:"owner_id"`
    Status    string            `json:"status"`
    Metadata  map[string]string `json:"metadata,omitempty"`
    CreatedAt int64             `json:"created_at"`
}

func main() {
    // 创建测试数据
    project := Project{
        ID:        "proj-12345",
        Name:      "AI IDE Backend",
        OwnerID:   "user-789",
        Status:    "active",
        Metadata:  map[string]string{"env": "production", "region": "us-east-1"},
        CreatedAt: time.Now().Unix(),
    }
    
    // 不同JSON库的基准测试
    iterations := 100000
    
    // 标准库
    start := time.Now()
    for i := 0; i < iterations; i++ {
        json.Marshal(project)
    }
    stdlibTime := time.Since(start)
    
    // go-json (无反射)
    start = time.Now()
    for i := 0; i < iterations; i++ {
        go_json.Marshal(project)
    }
    goJsonTime := time.Since(start)
    
    // json-iterator (动态适配)
    jsoniter := jsoniter.ConfigCompatibleWithStandardLibrary
    start = time.Now()
    for i := 0; i < iterations; i++ {
        jsoniter.Marshal(project)
    }
    jsoniterTime := time.Since(start)
    
    fmt.Printf("标准库:     %v\n", stdlibTime)
    fmt.Printf("go-json:   %v (%.2fx)\n", goJsonTime, float64(stdlibTime)/float64(goJsonTime))
    fmt.Printf("json-iterator: %v (%.2fx)\n", jsoniterTime, float64(stdlibTime)/float64(jsoniterTime))
}

package main

import (
    "context"
    "fmt"
    "log"
    "time"
    
    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials/insecure"
    "google.golang.org/grpc/resolver"
    "google.golang.org/grpc/balancer/roundrobin"
    
    pb "github.com/aiide/backend/gen/go/project/v1"
)

// 项目服务客户端封装
type ProjectServiceClient struct {
    conn *grpc.ClientConn
    client pb.ProjectServiceClient
}

// 初始化项目服务客户端（使用round_robin负载均衡）
func NewProjectServiceClient(cc *grpc.ClientConn) *ProjectServiceClient {
    return &ProjectServiceClient{
        conn:   cc,
        client: pb.NewProjectServiceClient(cc),
    }
}

// 创建项目
func (c *ProjectServiceClient) CreateProject(ctx context.Context, req *pb.CreateProjectRequest) (*pb.CreateProjectResponse, error) {
    ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
    defer cancel()
    return c.client.CreateProject(ctx, req)
}

// 获取项目
func (c *ProjectServiceClient) GetProject(ctx context.Context, req *pb.GetProjectRequest) (*pb.GetProjectResponse, error) {
    ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
    defer cancel()
    return c.client.GetProject(ctx, req)
}

// 自定义DNS解析器（支持静态服务发现）
type staticResolver struct {
    addresses []resolver.Address
}

func (r *staticResolver) Scheme() string { return "static" }

func (r *staticResolver) ResolveNow(opts resolver.ResolveNowOptions) {}

func (r *staticResolver) Close() {}

func NewStaticResolver(addresses []string) resolver.Builder {
    addrs := make([]resolver.Address, len(addresses))
    for i, addr := range addresses {
        addrs[i] = resolver.Address{Addr: addr}
    }
    return &staticResolverBuilder{addresses: addrs}
}

type staticResolverBuilder struct {
    addresses []resolver.Address
}

func (b *staticResolverBuilder) Scheme() string { return "static" }

func (b *staticResolverBuilder) Build(
    target resolver.Target,
    cc resolver.ClientConn,
    opts resolver.BuildOptions,
) (resolver.Resolver, error) {
    r := &staticResolver{addresses: b.addresses}
    // 立即触发一次解析
    cc.UpdateState(resolver.State{Addresses: b.addresses})
    return r, nil
}

func main() {
    // 注册静态解析器
    resolver.Register(NewStaticResolver(""))
    
    // gRPC连接选项
    var opts []grpc.DialOption
    
    // 使用不安全凭证（生产环境应使用TLS）
    opts = append(opts, grpc.WithTransportCredentials(insecure.NewCredentials()))
    
    // 设置负载均衡策略为 round_robin
    opts = append(opts, grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy":"round_robin"}`))
    
    // 连接目标（使用自定义scheme）
    target := "static:///project-service"
    
    // 建立连接（多个服务端地址）
    conn, err := grpc.NewClient(
        target,
        opts...,
    )
    if err != nil {
        log.Fatalf("Failed to connect: %v", err)
    }
    defer conn.Close()
    
    // 创建客户端
    client := NewProjectServiceClient(conn)
    
    // 测试RPC调用
    ctx := context.Background()
    
    // 创建项目
    createResp, err := client.CreateProject(ctx, &pb.CreateProjectRequest{
        Name:        "test-project",
        Description: "A test project",
    })
    if err != nil {
        log.Printf("CreateProject failed: %v", err)
    } else {
        fmt.Printf("Created project: %s\n", createResp.Project.Id)
    }
    
    // 获取项目
    getResp, err := client.GetProject(ctx, &pb.GetProjectRequest{
        Id: "proj-123",
    })
    if err != nil {
        log.Printf("GetProject failed: %v", err)
    } else {
        fmt.Printf("Got project: %s\n", getResp.Project.Name)
    }
}

# Envoy配置示例 - aiide-execution服务
static_resources:
  listeners:
    - name: execution_listener
      address:
        socket_address:
          address: 0.0.0.0
          port_value: 15001
      filter_chains:
        - filters:
            - name: envoy.filters.network.http_connection_manager
              typed_config:
                "@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager
                codec_type: auto
                stat_prefix: execution_service
                route_config:
                  name: execution_route
                  virtual_hosts:
                    - name: execution_service
                      domains: ["*"]
                      routes:
                        - match: { prefix: "/" }
                          route:
                            cluster: execution_cluster
                            timeout: 300s
                http_filters:
                  - name: envoy.filters.http.router
                    typed_config:
                      "@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.Router
  
  clusters:
    - name: execution_cluster
      type: EDS  # Endpoint Discovery Service
      eds_cluster_config:
        service_name: execution-service
        ads_config:
          api_type: GRPC
          grpc_services:
            - envoy.config.cluster.discovery.v3.AggregatedClusterConfigService:
                cluster_name: xds_cluster
      lb_policy: LEAST_REQUEST  # 最小请求数负载均衡
      health_checks:
        - timeout: 5s
          interval: 10s
          unhealthy_threshold: 3
          healthy_threshold: 2
          grpc_health_check:
            service_name: aiide.execution.v1.ExecutionService

package balancer

import (
    "context"
    "math/rand"
    "sync"
    
    "google.golang.org/grpc/balancer"
    "google.golang.org/grpc/balancer/base"
    "google.golang.org/grpc/resolver"
)

// RegionAwarePicker 区域感知的负载均衡器
// 优先选择同区域的节点，如果同区域无可用节点再选择其他区域
type RegionAwarePicker struct {
    mu         sync.RWMutex
    addresses  []resolver.Address
    subConns   map[string]balancer.SubConn
    region     string  // 当前客户端所在区域
    
    // 区域统计
    regionStats map[string]*regionStat
}

type regionStat struct {
    requests int64
    failures int64
}

func (p *RegionAwarePicker) Pick(ctx context.Context, opts balancer.PickOptions) (balancer.SubConn, balancer.PickResult, error) {
    p.mu.RLock()
    defer p.mu.RUnlock()
    
    var sameRegionAddrs []resolver.Address
    var otherRegionAddrs []resolver.Address
    
    for _, addr := range p.addresses {
        addrRegion := addr.Attributes.Value("region").(string)
        if addrRegion == p.region {
            sameRegionAddrs = append(sameRegionAddrs, addr)
        } else {
            otherRegionAddrs = append(otherRegionAddrs, addr)
        }
    }
    
    var selectedAddr resolver.Address
    
    if len(sameRegionAddrs) > 0 {
        // 同区域优先，使用加权随机（基于健康状态）
        selectedAddr = p.weightedSelect(sameRegionAddrs)
    } else if len(otherRegionAddrs) > 0 {
        // 跨区域fallback
        selectedAddr = otherRegionAddrs[rand.Intn(len(otherRegionAddrs))]
    } else {
        return nil, balancer.PickResult{}, nil
    }
    
    subConn := p.subConns[selectedAddr.Addr]
    return balancer.PickResult{SubConn: subConn}, nil, nil
}

func (p *RegionAwarePicker) weightedSelect(addrs []resolver.Address) resolver.Address {
    // 简化实现：随机选择，实际生产应使用加权算法
    return addrs[rand.Intn(len(addrs))]
}

// 区域感知负载均衡器Builder
type RegionAwareBalancerBuilder struct{}

func (b *RegionAwareBalancerBuilder) Build(cc balancer.ClientConn, opts balancer.BuildOptions) balancer.Balancer {
    return &RegionAwarePicker{
        subConns:    make(map[string]balancer.SubConn),
        regionStats: make(map[string]*regionStat),
    }
}

func (b *RegionAwareBalancerBuilder) Name() string { return "region_aware" }

package main

import (
    "context"
    "fmt"
    "sync"
    "time"
    
    "google.golang.org/grpc"
    "google.golang.org/grpc/codes"
    "google.golang.org/grpc/status"
)

// CircuitBreaker 熔断器实现
type CircuitBreaker struct {
    name             string
    failureThreshold float64  // 失败率阈值 (0-1)
    successThreshold float64  // 半开转关闭的成功率阈值
    windowDuration   time.Duration
    openDuration     time.Duration
    minRequestVolume int
    
    mu              sync.RWMutex
    state           CircuitState
    failureCount    int64
    successCount    int64
    totalCount      int64
    lastFailureTime time.Time
    windowStart     time.Time
}

type CircuitState int

const (
    StateClosed CircuitState = iota
    StateHalfOpen
    StateOpen
)

const (
    DefaultFailureThreshold = 0.5
    DefaultSuccessThreshold = 0.3
    DefaultWindowDuration   = 10 * time.Second
    DefaultOpenDuration     = 60 * time.Second
    DefaultMinRequestVolume = 20
)

// NewCircuitBreaker 创建熔断器
func NewCircuitBreaker(name string, opts ...CircuitBreakerOption) *CircuitBreaker {
    cb := &CircuitBreaker{
        name:             name,
        failureThreshold: DefaultFailureThreshold,
        successThreshold: DefaultSuccessThreshold,
        windowDuration:   DefaultWindowDuration,
        openDuration:      DefaultOpenDuration,
        minRequestVolume:  DefaultMinRequestVolume,
        state:             StateClosed,
        windowStart:       time.Now(),
    }
    
    for _, opt := range opts {
        opt(cb)
    }
    
    return cb
}

// CircuitBreakerOption 熔断器配置选项
type CircuitBreakerOption func(*CircuitBreaker)

// WithFailureThreshold 设置失败率阈值
func WithFailureThreshold(threshold float64) CircuitBreakerOption {
    return func(cb *CircuitBreaker) {
        cb.failureThreshold = threshold
    }
}

// WithOpenDuration 设置熔断持续时间
func WithOpenDuration(d time.Duration) CircuitBreakerOption {
    return func(cb *CircuitBreaker) {
        cb.openDuration = d
    }
}

// Execute 执行带熔断保护的调用
func (cb *CircuitBreaker) Execute(ctx context.Context, fn func() error) error {
    // 检查熔断状态
    if !cb.allowRequest() {
        return status.Errorf(codes.Unavailable, 
            "circuit breaker is open for %s", cb.name)
    }
    
    // 执行调用
    err := fn()
    
    // 记录结果
    cb.recordResult(err)
    
    return err
}

func (cb *CircuitBreaker) allowRequest() bool {
    cb.mu.Lock()
    defer cb.mu.Unlock()
    
    switch cb.state {
    case StateClosed:
        return true
        
    case StateOpen:
        // 检查是否超过熔断时间
        if time.Since(cb.lastFailureTime) > cb.openDuration {
            cb.state = StateHalfOpen
            cb.resetCounters()
            return true
        }
        return false
        
    case StateHalfOpen:
        return true  // 半开状态下允许有限请求
    }
    
    return true
}

func (cb *CircuitBreaker) recordResult(err error) {
    cb.mu.Lock()
    defer cb.mu.Unlock()
    
    cb.totalCount++
    
    if err != nil {
        cb.failureCount++
        cb.lastFailureTime = time.Now()
    } else {
        cb.successCount++
    }
    
    cb.evaluateState()
}

func (cb *CircuitBreaker) evaluateState() {
    // 滑动窗口重置
    if time.Since(cb.windowStart) > cb.windowDuration {
        cb.resetCounters()
        cb.windowStart = time.Now()
    }
    
    switch cb.state {
    case StateClosed:
        // 检查失败率
        if cb.totalCount >= int64(cb.minRequestVolume) {
            failureRate := float64(cb.failureCount) / float64(cb.totalCount)
            if failureRate >= cb.failureThreshold {
                cb.state = StateOpen
                cb.lastFailureTime = time.Now()
            }
        }
        
    case StateHalfOpen:
        // 检查成功率
        if cb.successCount >= int64(cb.minRequestVolume) {
            successRate := float64(cb.successCount) / float64(cb.totalCount)
            if successRate >= cb.successThreshold {
                cb.state = StateClosed
                cb.resetCounters()
            }
        }
    }
}

func (cb *CircuitBreaker) resetCounters() {
    cb.failureCount = 0
    cb.successCount = 0
    cb.totalCount = 0
}

// GetState 获取当前熔断状态（用于监控）
func (cb *CircuitBreaker) GetState() CircuitState {
    cb.mu.RLock()
    defer cb.mu.RUnlock()
    return cb.state
}

// 熔断器注册表（管理所有熔断器）
type CircuitBreakerRegistry struct {
    breakers map[string]*CircuitBreaker
    mu       sync.RWMutex
}

var globalRegistry = &CircuitBreakerRegistry{
    breakers: make(map[string]*CircuitBreaker),
}

func GetCircuitBreaker(name string) *CircuitBreaker {
    globalRegistry.mu.RLock()
    cb, exists := globalRegistry.breakers[name]
    globalRegistry.mu.RUnlock()
    
    if exists {
        return cb
    }
    
    globalRegistry.mu.Lock()
    defer globalRegistry.mu.Unlock()
    
    if cb, exists = globalRegistry.breakers[name]; exists {
        return cb
    }
    
    cb = NewCircuitBreaker(name)
    globalRegistry.breakers[name] = cb
    return cb
}

package main

import (
    "context"
    "fmt"
    
    "github.com/aiide/backend/pkg/cache"
)

// FallbackExecutor 降级执行器
type FallbackExecutor struct {
    strategies map[string]FallbackStrategy
}

// FallbackStrategy 降级策略接口
type FallbackStrategy interface {
    Execute(ctx context.Context) (interface{}, error)
}

// GetProjectsFallback 获取项目列表的降级策略
type GetProjectsFallback struct {
    cache cache.Cache
    req   interface{}
}

func (f *GetProjectsFallback) Execute(ctx context.Context) (interface{}, error) {
    // 尝试从缓存获取
    cacheKey := fmt.Sprintf("projects:user:fallback")
    
    cached, err := f.cache.Get(ctx, cacheKey)
    if err == nil && cached != nil {
        return cached, nil
    }
    
    // 缓存未命中，返回空列表（而不是错误）
    return []interface{}{}, nil
}

// FallbackResponse 统一降级响应
type FallbackResponse struct {
    Data     interface{}
    FromCache bool
    Stale     bool
}

// ExecuteWithFallback 执行带降级的调用
func ExecuteWithFallback(
    ctx context.Context,
    breakerName string,
    primaryFn func() (interface{}, error),
    fallbackFn FallbackStrategy,
) (*FallbackResponse, error) {
    breaker := GetCircuitBreaker(breakerName)
    
    // 尝试主调用
    err := breaker.Execute(ctx, func() error {
        _, err := primaryFn()
        return err
    })
    
    if err == nil {
        result, err := primaryFn()
        return &FallbackResponse{Data: result, FromCache: false}, err
    }
    
    // 主调用失败，执行降级
    if fallbackFn != nil {
        result, err := fallbackFn.Execute(ctx)
        return &FallbackResponse{Data: result, FromCache: true, Stale: true}, err
    }
    
    return nil, err
}

package main

import (
    "context"
    "fmt"
    "net/http"
    "time"
    
    "google.golang.org/grpc"
    "google.golang.org/grpc/health/grpc_health_v1"
)

// HealthCheckServer 健康检查服务实现
type HealthCheckServer struct {
    registry *ServiceRegistry
}

func NewHealthCheckServer(registry *ServiceRegistry) *HealthCheckServer {
    return &HealthCheckServer{registry: registry}
}

// Check 实现gRPC健康检查接口
func (h *HealthCheckServer) Check(ctx context.Context, req *grpc_health_v1.HealthCheckRequest) (*grpc_health_v1.HealthCheckResponse, error) {
    serviceName := req.GetService()
    
    // 检查服务是否注册
    service := h.registry.Get(serviceName)
    if service == nil {
        return &grpc_health_v1.HealthCheckResponse{
            Status: grpc_health_v1.HealthCheckResponse_NOT_SERVING,
        }, nil
    }
    
    // 检查熔断器状态
    breaker := GetCircuitBreaker(serviceName)
    state := breaker.GetState()
    
    switch state {
    case StateOpen:
        return &grpc_health_v1.HealthCheckResponse{
            Status: grpc_health_v1.HealthCheckResponse_NOT_SERVING,
        }, nil
    case StateHalfOpen:
        return &grpc_health_v1.HealthCheckResponse{
            Status: grpc_health_v1.HealthCheckResponse_SERVICE_UNKNOWN,
        }, nil
    default:
        return &grpc_health_v1.HealthCheckResponse{
            Status: grpc_health_v1.HealthCheckResponse_SERVING,
        }, nil
    }
}

// Watch 实现gRPC健康检查Watch接口
func (h *HealthCheckServer) Watch(req *grpc_health_v1.HealthCheckRequest, stream grpc_health_v1.Health_WatchServer) error {
    // 定期推送健康状态
    ticker := time.NewTicker(5 * time.Second)
    defer ticker.Stop()
    
    serviceName := req.GetService()
    
    for {
        select {
        case <-stream.Context().Done():
            return nil
        case <-ticker.C:
            service := h.registry.Get(serviceName)
            if service == nil {
                stream.Send(&grpc_health_v1.HealthCheckResponse{
                    Status: grpc_health_v1.HealthCheckResponse_NOT_SERVING,
                })
                continue
            }
            
            breaker := GetCircuitBreaker(serviceName)
            state := breaker.GetState()
            
            var status grpc_health_v1.HealthCheckResponse_ServingStatus
            switch state {
            case StateOpen:
                status = grpc_health_v1.HealthCheckResponse_NOT_SERVING
            case StateHalfOpen:
                status = grpc_health_v1.HealthCheckResponse_CONNECTING
            default:
                status = grpc_health_v1.HealthCheckResponse_SERVING
            }
            
            stream.Send(&grpc_health_v1.HealthCheckResponse{
                Status: status,
            })
        }
    }
}

// HTTP健康检查端点（用于K8s探活）
type HealthHandler struct {
    registry *ServiceRegistry
}

func (h *HealthHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
    checks := make(map[string]string)
    
    for _, name := range []string{"project", "execution", "ai-inference"} {
        breaker := GetCircuitBreaker(name)
        state := breaker.GetState()
        
        switch state {
        case StateClosed:
            checks[name] = "healthy"
        case StateHalfOpen:
            checks[name] = "degraded"
        case StateOpen:
            checks[name] = "unhealthy"
        }
    }
    
    // 检查是否有unhealthy的服务
    allHealthy := true
    for _, status := range checks {
        if status == "unhealthy" {
            allHealthy = false
            break
        }
    }
    
    w.Header().Set("Content-Type", "application/json")
    if allHealthy {
        w.WriteHeader(http.StatusOK)
        fmt.Fprintf(w, `{"status":"healthy","checks":%v}`, checks)
    } else {
        w.WriteHeader(http.StatusServiceUnavailable)
        fmt.Fprintf(w, `{"status":"unhealthy","checks":%v}`, checks)
    }
}

package tracing

import (
    "context"
    "fmt"
    "time"
    
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/attribute"
    "go.opentelemetry.io/otel/codes"
    "go.opentelemetry.io/otel/propagation"
    "go.opentelemetry.io/otel/trace"
    
    "google.golang.org/grpc"
    "google.golang.org/grpc/metadata"
    "google.golang.org/grpc/status"
)

const (
    // TraceIDKey gRPC metadata中的TraceID key
    TraceIDKey = "x-trace-id"
    // SpanIDKey gRPC metadata中的SpanID key
    SpanIDKey = "x-span-id"
    // TraceparentHeader W3C Trace Context格式
    TraceparentHeader = "traceparent"
)

// UnaryServerInterceptor OpenTelemetry unary服务端拦截器
func UnaryServerInterceptor(serviceName string, opts ...Option) grpc.UnaryServerInterceptor {
    cfg := &config{
        tracer:        otel.Tracer(serviceName),
        propagator:    otel.GetTextMapPropagator(),
        recorder:      DefaultRecorder{},
    }
    
    for _, opt := range opts {
        opt(cfg)
    }
    
    return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
        // 从gRPC Metadata提取Trace Context
        md, ok := metadata.FromIncomingContext(ctx)
        if !ok {
            md = metadata.MD{}
        }
        
        // 使用propagator提取上下文
        ctx = cfg.propagator.Extract(ctx, propagation.HeaderCarrier(md))
        
        // 创建新的span
        spanName := fmt.Sprintf("%s/%s", serviceName, info.FullMethod)
        ctx, span := cfg.tracer.Start(ctx, spanName,
            trace.WithSpanKind(trace.SpanKindServer),
            trace.WithAttributes(
                attribute.String("rpc.system", "grpc"),
                attribute.String("rpc.service", serviceName),
                attribute.String("rpc.method", info.FullMethod),
            ),
        )
        defer span.End()
        
        // 记录开始时间
        startTime := time.Now()
        
        // 调用处理函数
        resp, err := handler(ctx, req)
        
        // 记录结果
        duration := time.Since(startTime)
        recordSpan(ctx, span, err, duration, cfg.recorder)
        
        return resp, err
    }
}

// StreamServerInterceptor OpenTelemetry流式服务端拦截器
func StreamServerInterceptor(serviceName string, opts ...Option) grpc.StreamServerInterceptor {
    cfg := &config{
        tracer:     otel.Tracer(serviceName),
        propagator: otel.GetTextMapPropagator(),
        recorder:   DefaultRecorder{},
    }
    
    for _, opt := range opts {
        opt(cfg)
    }
    
    return func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
        // 从流上下文提取Trace Context
        md, ok := metadata.FromIncomingContext(ss.Context())
        if !ok {
            md = metadata.MD{}
        }
        
        ctx := cfg.propagator.Extract(ss.Context(), propagation.HeaderCarrier(md))
        
        // 创建新的span
        spanName := fmt.Sprintf("%s/%s", serviceName, info.FullMethod)
        ctx, span := cfg.tracer.Start(ctx, spanName,
            trace.WithSpanKind(trace.SpanKindServer),
            trace.WithAttributes(
                attribute.String("rpc.system", "grpc"),
                attribute.String("rpc.service", serviceName),
                attribute.String("rpc.method", info.FullMethod),
                attribute.Bool("rpc.streaming", info.IsServerStream),
            ),
        )
        defer span.End()
        
        // 用新的context包装流
        wrapped := &serverStreamWrapper{
            ServerStream: ss,
            ctx:          ctx,
        }
        
        // 调用处理函数
        err := handler(srv, wrapped)
        
        recordSpan(ctx, span, err, 0, cfg.recorder)
        
        return err
    }
}

type serverStreamWrapper struct {
    grpc.ServerStream
    ctx context.Context
}

func (w *serverStreamWrapper) Context() context.Context {
    return w.ctx
}

func recordSpan(ctx context.Context, span trace.Span, err error, duration time.Duration, recorder Recorder) {
    if err != nil {
        span.SetStatus(codes.Error, err.Error())
        span.RecordError(err)
        
        s, _ := status.FromError(err)
        span.SetAttributes(
            attribute.Int("rpc.grpc.status_code", int(s.Code())),
        )
    } else {
        span.SetStatus(codes.Ok, "")
        span.SetAttributes(
            attribute.Int("rpc.grpc.status_code", 0),
        )
    }
    
    if duration > 0 {
        span.SetAttributes(attribute.Int64("rpc.duration_ms", duration.Milliseconds()))
    }
    
    // 如果配置了recorder，记录到存储
    if recorder != nil {
        recorder.Record(ctx, span)
    }
}

package tracing

import (
    "context"
    "time"
    
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/attribute"
    "go.opentelemetry.io/otel/codes"
    "go.opentelemetry.io/otel/propagation"
    "go.opentelemetry.io/otel/trace"
    
    "google.golang.org/grpc"
    "google.golang.org/grpc/metadata"
    "google.golang.org/grpc/status"
)

// UnaryClientInterceptor OpenTelemetry unary客户端拦截器
func UnaryClientInterceptor(serviceName string, opts ...Option) grpc.UnaryClientInterceptor {
    cfg := &config{
        tracer:     otel.Tracer(serviceName),
        propagator: otel.GetTextMapPropagator(),
    }
    
    for _, opt := range opts {
        opt(cfg)
    }
    
    return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
        // 从当前context提取或创建Trace Context
        spanName := fmt.Sprintf("%s/%s", serviceName, method)
        ctx, span := cfg.tracer.Start(ctx, spanName,
            trace.WithSpanKind(trace.SpanKindClient),
            trace.WithAttributes(
                attribute.String("rpc.system", "grpc"),
                attribute.String("rpc.service", serviceName),
                attribute.String("rpc.method", method),
            ),
        )
        defer span.End()
        
        // 将Trace Context注入到gRPC Metadata
        md, ok := metadata.FromOutgoingContext(ctx)
        if !ok {
            md = metadata.MD{}
        }
        
        // 使用propagator注入上下文
        cfg.propagator.Inject(ctx, propagation.HeaderCarrier(md))
        
        // 将修改后的metadata放入context
        ctx = metadata.NewOutgoingContext(ctx, md)
        
        // 记录开始时间
        startTime := time.Now()
        
        // 执行RPC调用
        err := invoker(ctx, method, req, reply, cc, opts...)
        
        // 记录结果
        duration := time.Since(startTime)
        if err != nil {
            span.SetStatus(codes.Error, err.Error())
            span.RecordError(err)
            
            s, _ := status.FromError(err)
            span.SetAttributes(
                attribute.Int("rpc.grpc.status_code", int(s.Code())),
            )
        } else {
            span.SetStatus(codes.Ok, "")
            span.SetAttributes(
                attribute.Int("rpc.grpc.status_code", 0),
            )
        }
        
        span.SetAttributes(attribute.Int64("rpc.duration_ms", duration.Milliseconds()))
        
        return err
    }
}

// StreamClientInterceptor OpenTelemetry流式客户端拦截器
func StreamClientInterceptor(serviceName string, opts ...Option) grpc.StreamClientInterceptor {
    cfg := &config{
        tracer:     otel.Tracer(serviceName),
        propagator: otel.GetTextMapPropagator(),
    }
    
    for _, opt := range opts {
        opt(cfg)
    }
    
    return func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
        // 创建span
        spanName := fmt.Sprintf("%s/%s", serviceName, method)
        ctx, span := cfg.tracer.Start(ctx, spanName,
            trace.WithSpanKind(trace.SpanKindClient),
            trace.WithAttributes(
                attribute.String("rpc.system", "grpc"),
                attribute.String("rpc.service", serviceName),
                attribute.String("rpc.method", method),
                attribute.Bool("rpc.streaming", desc.ServerStreams),
            ),
        )
        defer span.End()
        
        // 注入Trace Context
        md, ok := metadata.FromOutgoingContext(ctx)
        if !ok {
            md = metadata.MD{}
        }
        
        cfg.propagator.Inject(ctx, propagation.HeaderCarrier(md))
        ctx = metadata.NewOutgoingContext(ctx, md)
        
        // 创建流
        stream, err := streamer(ctx, desc, cc, method, opts...)
        if err != nil {
            span.RecordError(err)
            span.SetStatus(codes.Error, err.Error())
            return nil, err
        }
        
        // 返回包装后的流
        return &clientStreamWrapper{
            ClientStream: stream,
            ctx:          ctx,
            span:         span,
        }, nil
    }
}

type clientStreamWrapper struct {
    grpc.ClientStream
    ctx  context.Context
    span trace.Span
}

func (w *clientStreamWrapper) Context() context.Context {
    return w.ctx
}

func (w *clientStreamWrapper) SendMsg(m interface{}) error {
    err := w.ClientStream.SendMsg(m)
    if err != nil {
        w.span.RecordError(err)
    }
    return err
}

func (w *clientStreamWrapper) RecvMsg(m interface{}) error {
    err := w.ClientStream.RecvMsg(m)
    if err != nil {
        w.span.RecordError(err)
    }
    return err
}

package main

import (
    "context"
    "fmt"
    
    "go.opentelemetry.io/otel/propagation"
    "google.golang.org/grpc"
    "google.golang.org/grpc/metadata"
)

// TraceContext传播示例
// 展示如何在服务间手动传播TraceID

type TracingInterceptor struct {
    propagator propagation.TextMapPropagator
}

func NewTracingInterceptor() *TracingInterceptor {
    return &TracingInterceptor{
        propagator: otel.GetTextMapPropagator(),
    }
}

// InjectTraceContext 将Trace上下文注入到context
func (t *TracingInterceptor) InjectTraceContext(ctx context.Context, md *metadata.MD) context.Context {
    // 从当前context提取
    carrier := propagation.HeaderCarrier{}
    ctx = t.propagator.Inject(ctx, carrier)
    
    // carrier现在包含了所有trace相关的header
    // 手动复制到gRPC metadata
    for k, v := range carrier.Get() {
        (*md)[k] = v
    }
    
    return ctx
}

// ExtractTraceContext 从gRPC metadata提取trace上下文
func (t *TracingInterceptor) ExtractTraceContext(ctx context.Context, md metadata.MD) context.Context {
    carrier := propagation.HeaderCarrier(md)
    return t.propagator.Extract(ctx, carrier)
}

// 服务间调用的Trace传播
func callDownstreamService(ctx context.Context, serviceName string, req interface{}) error {
    // 假设使用grpc.DialContext连接下游服务
    conn, err := grpc.DialContext(ctx, serviceName, grpc.WithInsecure())
    if err != nil {
        return err
    }
    defer conn.Close()
    
    // 从当前context获取metadata
    md, ok := metadata.FromIncomingContext(ctx)
    if !ok {
        md = metadata.MD{}
    }
    
    // 创建下游调用的metadata，复制原有metadata
    outMD := md.Copy()
    
    // 将trace context注入
    // 这里实际上是通过propagator自动完成的
    ctx = metadata.NewOutgoingContext(ctx, outMD)
    
    // 创建gRPC客户端并调用
    // ...
    
    fmt.Printf("TraceID propagated via metadata\n")
    return nil
}

# 安装Protocol Buffers编译器
# Linux/macOS
brew install protobuf
# 或从源码编译
git clone https://github.com/protocolbuffers/protobuf.git
cd protobuf
./autogen.sh
./configure
make -j$(nproc)
sudo make install

# Windows
choco install protobuf

# 验证安装
protoc --version  # libprotoc 3.21.x

# 安装Go插件
go install google.golang.org/protobuf/cmd/protoc-gen-go@latest
go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest

# 安装其他语言插件
# Java
#   1. 下载protoc-gen-java: https://repo1.maven.org/maven2/com/google/protobuf/protoc-gen/
#   2. 或使用Maven插件: protobuf-maven-plugin
#
# Python
pip install grpcio-tools
#
# JavaScript/TypeScript
npm install @grpc/proto-loader @types/protobufjs

#!/bin/bash
# generate_all.sh - 为所有语言生成gRPC代码

PROTO_DIR="./proto"
OUT_DIR="./generated"

# Go
protoc \
    --go_out=${OUT_DIR}/go \
    --go_opt=paths=source_relative \
    --go-grpc_out=${OUT_DIR}/go \
    --go-grpc_opt=paths=source_relative \
    -I${PROTO_DIR} \
    ${PROTO_DIR}/*.proto

# Python
python -m grpc_tools.protoc \
    --python_out=${OUT_DIR}/python \
    --grpc_python_out=${OUT_DIR}/python \
    -I${PROTO_DIR} \
    -I${OUT_DIR}/python \
    ${PROTO_DIR}/*.proto

# JavaScript (使用grpc-tools)
grpc_tools_node_protoc \
    --js_out=import_style=commonjs,binary:${OUT_DIR}/js \
    --grpc_out=grpc_js:${OUT_DIR}/js \
    --plugin=protoc-gen-grpc=./node_modules/.bin/grpc_tools_node_protoc_plugin \
    -I${PROTO_DIR} \
    ${PROTO_DIR}/*.proto

# python_client/project_client.py
"""
Python gRPC客户端实现
"""
import grpc
from typing import Iterator, Optional

import aiide.project.v1.project_pb2 as project_pb2
import aiide.project.v1.project_pb2_grpc as project_pb2_grpc


class ProjectServiceClient:
    """项目服务Python客户端"""
    
    def __init__(
        self,
        host: str = "localhost",
        port: int = 50051,
        timeout: float = 30.0,
        metadata: Optional[list] = None
    ):
        """
        初始化客户端连接
        
        Args:
            host: 服务地址
            port: gRPC端口
            timeout: 默认超时时间（秒）
            metadata: gRPC metadata（如认证token）
        """
        self.channel = grpc.insecure_channel(
            f"{host}:{port}",
            options=[
                ('grpc.max_send_message_length', 50 * 1024 * 1024),
                ('grpc.max_receive_message_length', 50 * 1024 * 1024),
                ('grpc.keepalive_time_ms', 30000),
                ('grpc.keepalive_timeout_ms', 10000),
            ]
        )
        self.stub = project_pb2_grpc.ProjectServiceStub(self.channel)
        self.timeout = timeout
        self.metadata = metadata or []
    
    def create_project(
        self,
        name: str,
        description: str = "",
        collaborator_ids: list = None,
        metadata: dict = None
    ) -> project_pb2.Project:
        """创建新项目"""
        request = project_pb2.CreateProjectRequest(
            name=name,
            description=description,
            collaborator_ids=collaborator_ids or [],
        )
        
        if metadata:
            request.metadata.update(metadata)
        
        try:
            response = self.stub.CreateProject(
                request,
                timeout=self.timeout,
                metadata=self.metadata
            )
            return response.project
        except grpc.RpcError as e:
            print(f"CreateProject failed: code={e.code()}, details={e.details()}")
            raise
    
    def get_project(self, project_id: str) -> project_pb2.Project:
        """获取项目详情"""
        request = project_pb2.GetProjectRequest(id=project_id)
        
        try:
            response = self.stub.GetProject(
                request,
                timeout=self.timeout,
                metadata=self.metadata
            )
            return response.project
        except grpc.RpcError as e:
            print(f"GetProject failed: code={e.code()}, details={e.details()}")
            raise
    
    def list_user_projects(
        self,
        user_id: str,
        page_size: int = 20,
        page_token: str = ""
    ) -> tuple:
        """列出用户项目（分页）"""
        request = project_pb2.ListUserProjectsRequest(
            user_id=user_id,
            page_size=page_size,
            page_token=page_token
        )
        
        response = self.stub.ListUserProjects(
            request,
            timeout=self.timeout,
            metadata=self.metadata
        )
        
        return list(response.projects), response.next_page_token, response.total_count
    
    def watch_projects(self, project_ids: list) -> Iterator[project_pb2.ProjectChangeEvent]:
        """监听项目变更（流式）"""
        request = project_pb2.WatchProjectsRequest(project_ids=project_ids)
        
        try:
            for event in self.stub.WatchProjects(
                request,
                timeout=self.timeout,
                metadata=self.metadata
            ):
                yield event
        except grpc.RpcError as e:
            print(f"WatchProjects stream error: code={e.code()}, details={e.details()}")
            raise
    
    def close(self):
        """关闭连接"""
        self.channel.close()
    
    def __enter__(self):
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()


# 使用示例
if __name__ == "__main__":
    with ProjectServiceClient("localhost", 50051) as client:
        # 创建项目
        project = client.create_project(
            name="test-project",
            description="A test project",
            metadata={"env": "development"}
        )
        print(f"Created project: {project.id}, name={project.name}")
        
        # 获取项目
        fetched = client.get_project(project.id)
        print(f"Fetched project: {fetched.name}, status={fetched.status}")
        
        # 列出项目
        projects, next_token, total = client.list_user_projects("user-123")
        print(f"User has {total} projects, first page: {len(projects)}")

// project_service.proto
// 添加HTTP注释实现REST API

syntax = "proto3";

package aiide.project.v1;

import "google/api/annotations.proto";
import "google/protobuf/empty.proto";
import "google/protobuf/timestamp.proto";

// HTTP REST API注释
service ProjectServiceREST {
  // 创建项目 - 对应 POST /v1/projects
  rpc CreateProject(CreateProjectRequest) returns (CreateProjectResponse) {
    option (google.api.http) = {
      post: "/v1/projects"
      body: "*"
    };
    option (grpc.gateway.protoc_gen_openapiv2.options.OPENAPIV2Operation) = {
      summary: "创建新项目"
      description: "创建一个新的AI IDE项目"
      tags: "项目管理"
    };
  }
  
  // 获取项目 - 对应 GET /v1/projects/{id}
  rpc GetProject(GetProjectRequest) returns (GetProjectResponse) {
    option (google.api.http) = {
      get: "/v1/projects/{id}"
    };
  }
  
  // 更新项目 - 对应 PATCH /v1/projects/{id}
  rpc UpdateProject(UpdateProjectRequest) returns (UpdateProjectResponse) {
    option (google.api.http) = {
      patch: "/v1/projects/{id}"
      body: "*"
    };
  }
  
  // 删除项目 - 对应 DELETE /v1/projects/{id}
  rpc DeleteProject(DeleteProjectRequest) returns (DeleteProjectResponse) {
    option (google.api.http) = {
      delete: "/v1/projects/{id}"
    };
  }
  
  // 列出用户项目 - 对应 GET /v1/users/{user_id}/projects
  rpc ListUserProjects(ListUserProjectsRequest) returns (ListUserProjectsResponse) {
    option (google.api.http) = {
      get: "/v1/users/{user_id}/projects"
    };
  }
}

# 安装 protoc-gen-openapiv2
go install github.com/grpc-ecosystem/grpc-gateway/v2/protoc-gen-openapiv2@latest

# 生成OpenAPI JSON
protoc \
    --openapiv2_out=${OUT_DIR}/openapi \
    --openapiv2_opt=logtostderr=true \
    --openapiv2_opt=allow_delete_body=true \
    --openapiv2_opt=generate_unbound_methods=true \
    -I${PROTO_DIR} \
    -I${HOME}/go/pkg/mod/cache/ \
    ${PROTO_DIR}/*.proto

aiide-backend/
├── cmd/                          # 应用入口
│   ├── gateway/                  # API网关
│   ├── auth-service/             # 认证服务
│   ├── project-service/          # 项目服务
│   ├── execution-service/        # 执行引擎
│   └── ai-inference-service/     # AI推理服务
│
├── api/                          # API定义
│   └── aiide/
│       └── v1/                   # v1版本API
│           ├── auth.proto
│           ├── project.proto
│           ├── execution.proto
│           └── ai_inference.proto
│
├── gen/                          # 生成的代码
│   ├── go/                       # Go语言
│   ├── python/                   # Python语言
│   └── js/                       # JavaScript语言
│
├── pkg/                          # 内部包
│   ├── interceptor/              # gRPC拦截器
│   │   ├── tracing/             # 追踪
│   │   ├── auth/                 # 认证
│   │   └── ratelimit/            # 限流
│   ├── balancer/                 # 负载均衡
│   ├── circuitbreaker/           # 熔断器
│   └── tracing/                  # 分布式追踪
│
├── docker/                       # Docker配置
│   ├── base/                     # 基础镜像
│   └── services/                 # 服务镜像
│
└── docker-compose.yml            # 本地开发编排

// cmd/project-service/main.go
package main

import (
    "context"
    "flag"
    "fmt"
    "log"
    "net"
    "os"
    "os/signal"
    "syscall"
    "time"
    
    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials"
    "google.golang.org/grpc/keepalive"
    "google.golang.org/grpc/reflection"
    
    "github.com/aiide/backend/api/aiide/v1"
    "github.com/aiide/backend/pkg/interceptor/auth"
    "github.com/aiide/backend/pkg/interceptor/ratelimit"
    "github.com/aiide/backend/pkg/interceptor/tracing"
    "github.com/aiide/backend/pkg/resilience"
)

var (
    port        = flag.Int("port", 50051, "gRPC server port")
    certFile    = flag.String("cert", "", "TLS certificate file")
    keyFile     = flag.String("key", "", "TLS key file")
    enableTrace = flag.Bool("trace", false, "Enable OpenTelemetry tracing")
)

func main() {
    flag.Parse()
    
    // 创建gRPC服务器选项
    var opts []grpc.ServerOption
    
    // TLS配置（生产环境必须）
    if *certFile != "" && *keyFile != "" {
        creds, err := credentials.NewServerTLSFromFile(*certFile, *keyFile)
        if err != nil {
            log.Fatalf("Failed to load TLS credentials: %v", err)
        }
        opts = append(opts, grpc.Creds(creds))
    }
    
    // 添加拦截器链
    var unaryInterceptors []grpc.UnaryServerInterceptor
    var streamInterceptors []grpc.StreamServerInterceptor
    
    // 1. 追踪拦截器（最外层）
    if *enableTrace {
        unaryInterceptors = append(unaryInterceptors, 
            tracing.UnaryServerInterceptor("project-service"))
        streamInterceptors = append(streamInterceptors,
            tracing.StreamServerInterceptor("project-service"))
    }
    
    // 2. 认证拦截器
    unaryInterceptors = append(unaryInterceptors,
        auth.UnaryServerInterceptor(auth.WithJWKS("https://auth.aiide.com/.well-known/jwks.json")))
    
    // 3. 限流拦截器
    rateLimiter := ratelimit.NewTokenBucketLimiter(1000, 100) // 1000 QPS, burst 100
    unaryInterceptors = append(unaryInterceptors,
        ratelimit.UnaryServerInterceptor(rateLimiter))
    
    // 组合拦截器
    opts = append(opts, grpc.ChainUnaryInterceptor(unaryInterceptors...))
    opts = append(opts, grpc.ChainStreamInterceptor(streamInterceptors...))
    
    // KeepAlive配置
    opts = append(opts, grpc.KeepaliveParams(keepalive.ServerParameters{
        MaxConnectionIdle:     5 * time.Minute,
        MaxConnectionAge:       2 * time.Hour,
        MaxConnectionAgeGrace:  30 * time.Second,
        Time:                   10 * time.Second,
        Timeout:                3 * time.Second,
    }))
    
    // 创建gRPC服务器
    server := grpc.NewServer(opts...)
    
    // 注册服务
    aiide_v1.RegisterProjectServiceServer(server, NewProjectServer())
    
    // 注册反射服务（用于grpcurl等工具）
    reflection.Register(server)
    
    // 创建监听器
    lis, err := net.Listen("tcp", fmt.Sprintf(":%d", *port))
    if err != nil {
        log.Fatalf("Failed to listen: %v", err)
    }
    
    log.Printf("Starting project-service on port %d", *port)
    
    // 优雅关闭
    go func() {
        sigCh := make(chan os.Signal, 1)
        signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
        <-sigCh
        
        log.Println("Shutting down server...")
        server.GracefulStop()
    }()
    
    // 启动服务器
    if err := server.Serve(lis); err != nil {
        log.Fatalf("Server exited with error: %v", err)
    }
}

// ProjectServer 项目服务实现
type ProjectServer struct {
    aiide_v1.UnimplementedProjectServiceServer
    
    // 依赖注入
    store     ProjectStore
    cache     Cache
    publisher EventPublisher
}

// NewProjectServer 创建ProjectServer实例
func NewProjectServer() *ProjectServer {
    return &ProjectServer{
        store:     NewPostgresStore(),
        cache:     NewRedisCache(),
        publisher: NewKafkaPublisher(),
    }
}

// CreateProject 实现创建项目
func (s *ProjectServer) CreateProject(ctx context.Context, req *aiide_v1.CreateProjectRequest) (*aiide_v1.CreateProjectResponse, error) {
    // 1. 参数验证
    if req.Name == "" {
        return nil, status.Errorf(codes.InvalidArgument, "project name is required")
    }
    if len(req.Name) > 255 {
        return nil, status.Errorf(codes.InvalidArgument, "project name too long (max 255 chars)")
    }
    
    // 2. 获取当前用户（从auth拦截器注入的context）
    userID := auth.GetUserIDFromContext(ctx)
    if userID == "" {
        return nil, status.Error(codes.Unauthenticated, "user not authenticated")
    }
    
    // 3. 检查名称唯一性
    existing, err := s.store.GetByName(ctx, userID, req.Name)
    if err != nil && err != ErrNotFound {
        return nil, status.Errorf(codes.Internal, "failed to check name uniqueness: %v", err)
    }
    if existing != nil {
        return nil, status.Errorf(codes.AlreadyExists, "project with name '%s' already exists", req.Name)
    }
    
    // 4. 创建项目
    project := &aiide_v1.Project{
        Id:               generateID("proj"),
        Name:             req.Name,
        Description:      req.Description,
        OwnerId:          userID,
        CollaboratorIds:  req.CollaboratorIds,
        Status:           aiide_v1.ProjectStatus_PROJECT_STATUS_INITIALIZING,
        CreatedAt:        time.Now().Unix(),
        UpdatedAt:        time.Now().Unix(),
        Metadata:          req.Metadata,
    }
    
    if err := s.store.Create(ctx, project); err != nil {
        return nil, status.Errorf(codes.Internal, "failed to create project: %v", err)
    }
    
    // 5. 缓存项目（异步）
    go func() {
        if err := s.cache.Set(ctx, project.Id, project, 1*time.Hour); err != nil {
            log.Printf("Failed to cache project %s: %v", project.Id, err)
        }
    }()
    
    // 6. 发布项目创建事件
    go func() {
        event := &aiide_v1.ProjectChangeEvent{
            ProjectId:  project.Id,
            ChangeType: aiide_v1.ProjectChangeEvent_CHANGE_TYPE_CREATED,
            Project:    project,
            Timestamp:  time.Now().Unix(),
        }
        if err := s.publisher.Publish(ctx, "project-events", event); err != nil {
            log.Printf("Failed to publish project creation event: %v", err)
        }
    }()
    
    return &aiide_v1.CreateProjectResponse{Project: project}, nil
}

// GetProject 实现获取项目
func (s *ProjectServer) GetProject(ctx context.Context, req *aiide_v1.GetProjectRequest) (*aiide_v1.GetProjectResponse, error) {
    if req.Id == "" {
        return nil, status.Errorf(codes.InvalidArgument, "project id is required")
    }
    
    // 1. 尝试从缓存获取
    cached, err := s.cache.Get(ctx, req.Id)
    if err == nil && cached != nil {
        return &aiide_v1.GetProjectResponse{Project: cached.(*aiide_v1.Project)}, nil
    }
    
    // 2. 从存储获取
    project, err := s.store.Get(ctx, req.Id)
    if err == ErrNotFound {
        return nil, status.Errorf(codes.NotFound, "project '%s' not found", req.Id)
    }
    if err != nil {
        return nil, status.Errorf(codes.Internal, "failed to get project: %v", err)
    }
    
    // 3. 更新缓存
    go func() {
        s.cache.Set(ctx, project.Id, project, 1*time.Hour)
    }()
    
    return &aiide_v1.GetProjectResponse{Project: project}, nil
}

// ListUserProjects 实现列出用户项目（分页）
func (s *ProjectServer) ListUserProjects(req *aiide_v1.ListUserProjectsRequest, stream aiide_v1.ProjectService_ListUserProjectsServer) error {
    if req.UserId == "" {
        return status.Errorf(codes.InvalidArgument, "user_id is required")
    }
    
    pageSize := int(req.PageSize)
    if pageSize <= 0 {
        pageSize = 20
    }
    if pageSize > 100 {
        pageSize = 100
    }
    
    offset := 0
    if req.PageToken != "" {
        var err error
        offset, err = decodePageToken(req.PageToken)
        if err != nil {
            return status.Errorf(codes.InvalidArgument, "invalid page_token")
        }
    }
    
    // 获取总数
    total, err := s.store.CountByUser(stream.Context(), req.UserId)
    if err != nil {
        return status.Errorf(codes.Internal, "failed to count projects: %v", err)
    }
    
    // 分页获取
    var cursor string
    for {
        projects, nextCursor, err := s.store.ListByUser(stream.Context(), req.UserId, pageSize, offset)
        if err != nil {
            return status.Errorf(codes.Internal, "failed to list projects: %v", err)
        }
        
        for _, project := range projects {
            if err := stream.Send(&aiide_v1.ListUserProjectsResponse{
                Projects:         projects,
                NextPageToken:    nextCursor,
                TotalCount:       int32(total),
            }); err != nil {
                return err
            }
        }
        
        if nextCursor == "" {
            break
        }
        cursor = nextCursor
    }
    
    return nil
}

// WatchProjects 实现流式监听项目变更
func (s *ProjectServer) WatchProjects(req *aiide_v1.WatchProjectsRequest, stream aiide_v1.ProjectService_WatchProjectsServer) error {
    // 订阅项目变更事件
    events, err := s.publisher.Subscribe(stream.Context(), "project-events", req.ProjectIds)
    if err != nil {
        return status.Errorf(codes.Internal, "failed to subscribe to events: %v", err)
    }
    
    for {
        select {
        case <-stream.Context().Done():
            return stream.Context().Err()
        case event, ok := <-events:
            if !ok {
                return nil
            }
            if err := stream.Send(event.(*aiide_v1.ProjectChangeEvent)); err != nil {
                return err
            }
        }
    }
}

// 辅助函数
func generateID(prefix string) string {
    return fmt.Sprintf("%s-%s", prefix, uuid.New().String())
}

func contains(slice []string, item string) bool {
    for _, s := range slice {
        if s == item {
            return true
        }
    }
    return false
}

func decodePageToken(token string) (int, error) {
    // 简化的page token实现，实际应使用加密或压缩
    return strconv.Atoi(token)
}

# docker/project-service/Dockerfile
FROM golang:1.22-alpine AS builder

# 安装构建依赖
RUN apk add --no-cache git ca-certificates

WORKDIR /app

# 复制go mod文件
COPY go.mod go.sum ./
RUN go mod download

# 复制源代码
COPY . .

# 构建
ARG SERVICE_NAME=project-service
RUN CGO_ENABLED=0 GOOS=linux go build \
    -ldflags="-s -w" \
    -o /app/bin/${SERVICE_NAME} \
    ./cmd/${SERVICE_NAME}

# 运行镜像
FROM alpine:3.19

RUN apk add --no-cache ca-certificates tzdata

# 创建非root用户
RUN addgroup -g 1000 appgroup && \
    adduser -u 1000 -G appgroup -s /bin/sh -D appuser

WORKDIR /app

# 从builder复制二进制文件
COPY --from=builder /app/bin/project-service .
COPY --from=builder /app/api /app/api

# 复制配置文件（如果需要）
COPY docker/project-service/config.yaml /app/config.yaml

# 切换到非root用户
USER appuser

# 健康检查
HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
    CMD wget --no-verbose --tries=1 --spider http://localhost:50051/healthz || exit 1

EXPOSE 50051

ENTRYPOINT ["./project-service"]
CMD ["--config", "/app/config.yaml"]

# docker-compose.yml
version: '3.8'

services:
  # API网关
  gateway:
    build:
      context: .
      dockerfile: docker/gateway/Dockerfile
    ports:
      - "8080:8080"
      - "50051:50051"
    environment:
      - GRPC_ENDPOINT=project-service:50051
      - AUTH_SERVICE=auth-service:50051
    depends_on:
      - project-service
      - auth-service
    networks:
      - aiide-network

  # 项目服务（多实例）
  project-service:
    build:
      context: .
      dockerfile: docker/project-service/Dockerfile
    deploy:
      replicas: 3
    ports:
      - "50051"
    environment:
      - DATABASE_URL=postgres://aiide:password@postgres:5432/aiide
      - REDIS_URL=redis://redis:6379
      - KAFKA_BROKERS=kafka:9092
    depends_on:
      - postgres
      - redis
      - kafka
    networks:
      - aiide-network
    healthcheck:
      test: ["CMD", "wget", "--no-verbose", "--tries=1", "--spider", "http://localhost:50051/healthz"]
      interval: 30s
      timeout: 3s
      retries: 3

  # 认证服务
  auth-service:
    build:
      context: .
      dockerfile: docker/auth-service/Dockerfile
    ports:
      - "50052"
    environment:
      - JWT_SECRET=${JWT_SECRET}
      - JWKS_URL=https://auth.aiide.com/.well-known/jwks.json
    networks:
      - aiide-network

  # 执行引擎服务
  execution-service:
    build:
      context: .
      dockerfile: docker/execution-service/Dockerfile
    deploy:
      replicas: 5
    ports:
      - "50053"
    environment:
      - DOCKER_HOST=tcp://docker-socket-proxy:2375
      - REDIS_URL=redis://redis:6379
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock  # 仅开发环境
    networks:
      - aiide-network

  # AI推理服务
  ai-inference-service:
    build:
      context: .
      dockerfile: docker/ai-inference-service/Dockerfile
    deploy:
      replicas: 2
    ports:
      - "50054"
    environment:
      - MODEL_PATH=/models
      - CUDA_VISIBLE_DEVICES=0,1
    volumes:
      - model-cache:/models
    networks:
      - aiide-network

  # 数据存储
  postgres:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: aiide
      POSTGRES_PASSWORD: password
      POSTGRES_DB: aiide
    volumes:
      - postgres-data:/var/lib/postgresql/data
    ports:
      - "5432:5432"
    networks:
      - aiide-network

  redis:
    image: redis:7-alpine
    command: redis-server --appendonly yes
    volumes:
      - redis-data:/data
    ports:
      - "6379:6379"
    networks:
      - aiide-network

  kafka:
    image: confluentinc/cp-kafka:7.5.0
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
    networks:
      - aiide-network

  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
    networks:
      - aiide-network

networks:
  aiide-network:
    driver: bridge

volumes:
  postgres-data:
  redis-data:
  model-cache:

package main

import (
    "context"
    "sync"
    "time"
    
    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials/insecure"
    "google.golang.org/grpc/keepalive"
)

// ConnectionPool gRPC连接池
type ConnectionPool struct {
    target       string
    dialOpts     []grpc.DialOption
    connections  []*grpc.ClientConn
    mu           sync.RWMutex
    index        int
    poolSize     int
}

// NewConnectionPool 创建连接池
func NewConnectionPool(target string, poolSize int, opts ...grpc.DialOption) (*ConnectionPool, error) {
    defaultOpts := []grpc.DialOption{
        grpc.WithTransportCredentials(insecure.NewCredentials()),
        grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy":"round_robin"}`),
        grpc.WithKeepaliveParams(keepalive.ClientParameters{
            Time:                20 * time.Second,
            Timeout:             10 * time.Second,
            PermitWithoutStream: true,
        }),
        // 连接级别的流量窗口
        grpc.WithInitialWindowSize(65535),
        grpc.WithInitialConnWindowSize(65535),
    }
    
    defaultOpts = append(defaultOpts, opts...)
    
    pool := &ConnectionPool{
        target:   target,
        dialOpts: defaultOpts,
        poolSize: poolSize,
    }
    
    // 预热连接
    if err := pool.warmup(context.Background()); err != nil {
        return nil, err
    }
    
    return pool, nil
}

func (p *ConnectionPool) warmup(ctx context.Context) error {
    var wg sync.WaitGroup
    errCh := make(chan error, p.poolSize)
    
    for i := 0; i < p.poolSize; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            
            conn, err := grpc.DialContext(ctx, p.target, p.dialOpts...)
            if err != nil {
                errCh <- err
                return
            }
            
            p.mu.Lock()
            p.connections = append(p.connections, conn)
            p.mu.Unlock()
        }()
    }
    
    wg.Wait()
    close(errCh)
    
    for err := range errCh {
        if err != nil {
            return err
        }
    }
    
    return nil
}

// Get 获取一个连接（轮询）
func (p *ConnectionPool) Get() *grpc.ClientConn {
    p.mu.Lock()
    defer p.mu.Unlock()
    
    conn := p.connections[p.index]
    p.index = (p.index + 1) % p.poolSize
    return conn
}

// Close 关闭所有连接
func (p *ConnectionPool) Close() {
    p.mu.Lock()
    defer p.mu.Unlock()
    
    for _, conn := range p.connections {
        conn.Close()
    }
    p.connections = nil
}

package metrics

import (
    "context"
    "strconv"
    "time"
    
    "github.com/prometheus/client_golang/prometheus"
    "github.com/prometheus/client_golang/prometheus/promauto"
    "google.golang.org/grpc"
    "google.golang.org/grpc/status"
)

// gRPC指标定义
var (
    // 服务级别指标
    requestsTotal = promauto.NewCounterVec(
        prometheus.CounterOpts{
            Name: "aiide_grpc_requests_total",
            Help: "Total number of gRPC requests",
        },
        []string{"service", "method", "status"},
    )
    
    requestDuration = promauto.NewHistogramVec(
        prometheus.HistogramOpts{
            Name:    "aiide_grpc_request_duration_seconds",
            Help:    "gRPC request duration in seconds",
            Buckets: prometheus.DefBuckets,
        },
        []string{"service", "method"},
    )
    
    // 连接级别指标
    activeConnections = promauto.NewGauge(
        prometheus.GaugeOpts{
            Name: "aiide_grpc_active_connections",
            Help: "Number of active gRPC connections",
        },
    )
    
    // 流式RPC指标
    streamingRPCs = promauto.NewCounterVec(
        prometheus.CounterOpts{
            Name: "aiide_grpc_streaming_rpcs_total",
            Help: "Total number of streaming RPCs",
        },
        []string{"service", "method", "type"},  // type: server-streaming, client-streaming, bidirectional
    )
    
    // 熔断器指标
    circuitBreakerState = promauto.NewGaugeVec(
        prometheus.GaugeOpts{
            Name: "aiide_circuit_breaker_state",
            Help: "Circuit breaker state (0=closed, 1=half-open, 2=open)",
        },
        []string{"service", "name"},
    )
    
    circuitBreakerFailures = promauto.NewCounterVec(
        prometheus.CounterOpts{
            Name: "aiide_circuit_breaker_failures_total",
            Help: "Total number of circuit breaker recorded failures",
        },
        []string{"service", "name"},
    )
)

// MetricsServerInterceptor gRPC服务指标拦截器
func MetricsServerInterceptor(serviceName string) grpc.UnaryServerInterceptor {
    return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
        start := time.Now()
        
        resp, err := handler(ctx, req)
        
        duration := time.Since(start).Seconds()
        statusCode := status.Code(err).String()
        
        requestsTotal.WithLabelValues(serviceName, info.FullMethod, statusCode).Inc()
        requestDuration.WithLabelValues(serviceName, info.FullMethod).Observe(duration)
        
        return resp, err
    }
}

// PrometheusExporter Prometheus导出器
type PrometheusExporter struct {
    registry *prometheus.Registry
}

// NewPrometheusExporter 创建Prometheus导出器
func NewPrometheusExporter() *PrometheusExporter {
    return &PrometheusExporter{
        registry: prometheus.NewRegistry(),
    }
}

// Register 将自定义指标注册到导出器
func (e *PrometheusExporter) Register() error {
    e.registry.MustRegister(requestsTotal)
    e.registry.MustRegister(requestDuration)
    e.registry.MustRegister(activeConnections)
    e.registry.MustRegister(circuitBreakerState)
    return nil
}

package log

import (
    "context"
    "encoding/json"
    "fmt"
    "time"
)

// LogEntry 标准日志格式
type LogEntry struct {
    Timestamp   string                 `json:"timestamp"`
    Level       string                 `json:"level"`
    Message     string                 `json:"message"`
    Service     string                 `json:"service"`
    Version     string                 `json:"version"`
    TraceID     string                 `json:"trace_id,omitempty"`
    SpanID      string                 `json:"span_id,omitempty"`
    UserID      string                 `json:"user_id,omitempty"`
    RequestID   string                 `json:"request_id,omitempty"`
    Duration    string                 `json:"duration,omitempty"`
    StatusCode  int                    `json:"status_code,omitempty"`
    Error       string                 `json:"error,omitempty"`
    Fields      map[string]interface{} `json:"fields,omitempty"`
}

// GrpcLogger gRPC日志记录器
type GrpcLogger struct {
    logger   Logger
    service  string
    traceID  string
    spanID   string
}

// Logger 日志接口
type Logger interface {
    Info(msg string, fields map[string]interface{})
    Error(msg string, fields map[string]interface{})
}

// NewGrpcLogger 创建gRPC日志记录器
func NewGrpcLogger(service string) *GrpcLogger {
    return &GrpcLogger{
        service: service,
    }
}

// WithContext 从context提取trace信息
func (l *GrpcLogger) WithContext(ctx context.Context) *GrpcLogger {
    traceID, spanID := getTraceIDAndSpanID(ctx)
    return &GrpcLogger{
        logger:  l.logger,
        service: l.service,
        traceID: traceID,
        spanID:  spanID,
    }
}

// getTraceIDAndSpanID 从context获取trace信息
func getTraceIDAndSpanID(ctx context.Context) (string, string) {
    // 这里应该集成OpenTelemetry的context提取
    return "", ""
}

// LogUnaryRPC 记录Unary RPC调用
func (l *GrpcLogger) LogUnaryRPC(method string, duration time.Duration, statusCode int, err error) {
    entry := LogEntry{
        Timestamp:  time.Now().UTC().Format(time.RFC3339),
        Level:      "info",
        Message:     fmt.Sprintf("gRPC unary call: %s", method),
        Service:    l.service,
        TraceID:    l.traceID,
        Duration:   duration.String(),
        StatusCode: statusCode,
    }
    
    if err != nil {
        entry.Level = "error"
        entry.Error = err.Error()
    }
    
    l.emit(entry)
}

// LogStreamRPC 记录流式RPC调用
func (l *GrpcLogger) LogStreamRPC(method string, streamType string, duration time.Duration, msgCount int, err error) {
    entry := LogEntry{
        Timestamp: time.Now().UTC().Format(time.RFC3339),
        Level:     "info",
        Message:   fmt.Sprintf("gRPC stream call: %s (%s)", method, streamType),
        Service:   l.service,
        TraceID:   l.traceID,
        Duration:  duration.String(),
        Fields: map[string]interface{}{
            "message_count": msgCount,
            "stream_type":   streamType,
        },
    }
    
    if err != nil {
        entry.Level = "error"
        entry.Error = err.Error()
    }
    
    l.emit(entry)
}

func (l *GrpcLogger) emit(entry LogEntry) {
    data, _ := json.Marshal(entry)
    fmt.Println(string(data))
}

// 下一代AI服务通信协议
message AIStreamRequest {
    string session_id = 1;
    string model_id = 2;
    repeated Message messages = 3;
    GenerationConfig config = 4;
    
    // 流式控制
    bool enable_partial_response = 5;  // 实时流式输出
    int32 partial_response_interval_ms = 6;
}

message AIStreamResponse {
    oneof event {
        TokenEvent token = 1;           // 逐token输出
        ProgressEvent progress = 2;    // 进度更新
        CompletionEvent completion = 3; // 最终结果
        ErrorEvent error = 4;          // 错误事件
    }
}

message TokenEvent {
    string token = 1;
    int32 token_count = 2;
    float log_probability = 3;
}

message CompletionEvent {
    string content = 1;
    int32 total_tokens = 2;
    string finish_reason = 3;
    UsageMetadata usage = 4;
}

Phase 1 (当前): 微服务架构
├── 现状: 基于gRPC的微服务群
├── 收益: 服务独立部署、独立扩展
└── 挑战: 服务治理复杂度高

Phase 2 (6-12月): 服务网格化
├── 目标: 引入Istio/Linkerd
├── 收益: 统一的服务治理、安全、可观测
└── 变化: 应用层逻辑下沉到基础设施

Phase 3 (12-24月): 混合云原生
├── 目标: 多云部署、边缘计算
├── 收益: 全球化低延迟、高可用
└── 变化: 服务拓扑动态调整

Phase 4 (长期): AI Native架构
├── 目标: 自适应计算、实时模型编排
├── 收益: 智能资源调度、最优用户体验
└── 变化: 从RPC到AI感知的通信协议

// aiide_service.proto - AI IDE服务集合定义
syntax = "proto3";

package aiide.v1;

import "google/api/annotations.proto";
import "google/protobuf/empty.proto";
import "google/protobuf/timestamp.proto";
import "google/protobuf/field_mask.proto";

// ========== 通用定义 ==========

// 分页请求
message ListRequest {
    int32 page_size = 1;
    string page_token = 2;
}

// 分页响应
message ListResponse {
    string next_page_token = 1;
    int32 total_size = 2;
}

// 通用状态
enum Status {
    STATUS_UNSPECIFIED = 0;
    STATUS_OK = 1;
    STATUS_ERROR = 2;
    STATUS_PENDING = 3;
}

// ========== 项目服务 ==========

message Project {
    string id = 1;
    string name = 2;
    string description = 3;
    string owner_id = 4;
    repeated string collaborator_ids = 5;
    ProjectStatus status = 6;
    map<string, string> metadata = 7;
    google.protobuf.Timestamp created_at = 8;
    google.protobuf.Timestamp updated_at = 9;
}

enum ProjectStatus {
    PROJECT_STATUS_UNSPECIFIED = 0;
    PROJECT_STATUS_INITIALIZING = 1;
    PROJECT_STATUS_ACTIVE = 2;
    PROJECT_STATUS_SUSPENDED = 3;
    PROJECT_STATUS_DELETED = 4;
}

message CreateProjectRequest {
    string name = 1;
    string description = 2;
    repeated string collaborator_ids = 3;
    map<string, string> metadata = 4;
}

message CreateProjectResponse {
    Project project = 1;
}

message GetProjectRequest {
    string id = 1;
}

message GetProjectResponse {
    Project project = 1;
}

message UpdateProjectRequest {
    string id = 1;
    string name = 2;
    string description = 3;
    ProjectStatus status = 4;
    map<string, string> metadata = 5;
    google.protobuf.FieldMask update_mask = 6;
}

message UpdateProjectResponse {
    Project project = 1;
}

message DeleteProjectRequest {
    string id = 1;
}

message DeleteProjectResponse {
    bool success = 1;
}

message ListUserProjectsRequest {
    string user_id = 1;
    int32 page_size = 2;
    string page_token = 3;
}

message ListUserProjectsResponse {
    repeated Project projects = 1;
    string next_page_token = 2;
    int32 total_size = 3;
}

message WatchProjectsRequest {
    repeated string project_ids = 1;
}

message WatchProjectsResponse {
    string project_id = 1;
    ChangeType change_type = 2;
    Project project = 3;
    google.protobuf.Timestamp timestamp = 4;
    
    enum ChangeType {
        CHANGE_TYPE_UNSPECIFIED = 0;
        CHANGE_TYPE_CREATED = 1;
        CHANGE_TYPE_UPDATED = 2;
        CHANGE_TYPE_DELETED = 3;
    }
}

service ProjectService {
    rpc CreateProject(CreateProjectRequest) returns (CreateProjectResponse) {
        option (google.api.http) = {
            post: "/v1/projects"
            body: "*"
        };
    }
    
    rpc GetProject(GetProjectRequest) returns (GetProjectResponse) {
        option (google.api.http) = {
            get: "/v1/projects/{id}"
        };
    }
    
    rpc UpdateProject(UpdateProjectRequest) returns (UpdateProjectResponse) {
        option (google.api.http) = {
            patch: "/v1/projects/{id}"
            body: "*"
        };
    }
    
    rpc DeleteProject(DeleteProjectRequest) returns (DeleteProjectResponse) {
        option (google.api.http) = {
            delete: "/v1/projects/{id}"
        };
    }
    
    rpc ListUserProjects(ListUserProjectsRequest) returns (ListUserProjectsResponse) {
        option (google.api.http) = {
            get: "/v1/users/{user_id}/projects"
        };
    }
    
    // 流式API
    rpc WatchProjects(WatchProjectsRequest) returns (stream WatchProjectsResponse);
}

// ========== 执行服务 ==========

message ExecutionRequest {
    string execution_id = 1;
    string language = 2;
    string code = 3;
    int32 timeout_ms = 4;
    int64 memory_limit_mb = 5;
    map<string, string> environment = 6;
    string stdin_input = 7;
}

message ExecutionResponse {
    string execution_id = 1;
    Status status = 2;
    string stdout = 3;
    string stderr = 4;
    int32 exit_code = 5;
    int64 execution_time_ms = 6;
    int64 memory_usage_mb = 7;
    string error_message = 8;
}

service ExecutionService {
    rpc Execute(ExecutionRequest) returns (ExecutionResponse);
    rpc ExecuteStream(ExecutionRequest) returns (stream ExecutionResponse);
    rpc CancelExecution(CancelExecutionRequest) returns (google.protobuf.Empty);
}

message CancelExecutionRequest {
    string execution_id = 1;
}

// ========== 健康检查服务 ==========

service HealthService {
    rpc Check(HealthCheckRequest) returns (HealthCheckResponse);
    rpc Watch(HealthCheckRequest) returns (stream HealthCheckResponse);
}

message HealthCheckRequest {
    string service = 1;
}

message HealthCheckResponse {
    string service = 1;
    ServingStatus status = 2;
    repeated HealthCheckResponse details = 3;
}

enum ServingStatus {
    SERVING_STATUS_UNSPECIFIED = 0;
    SERVING_STATUS_SERVING = 1;
    SERVING_STATUS_NOT_SERVING = 2;
    SERVING_STATUS_SERVICE_UNKNOWN = 3;
}

# python_client/complete_client.py
"""
AI IDE gRPC客户端完整实现
支持所有服务：Project、Execution、AI-Inference
"""

import grpc
import json
import time
from typing import Iterator, Optional, Callable
from contextlib import contextmanager

import aiide.project.v1.project_pb2 as project_pb2
import aiide.project.v1.project_pb2_grpc as project_pb2_grpc
import aiide.execution.v1.execution_pb2 as execution_pb2
import aiide.execution.v1.execution_pb2_grpc as execution_pb2_grpc


class AIIDEClient:
    """AI IDE统一客户端"""
    
    def __init__(
        self,
        host: str = "localhost",
        grpc_port: int = 50051,
        http_port: int = 8080,
        use_tls: bool = False,
        metadata: Optional[list] = None
    ):
        self.host = host
        self.grpc_port = grpc_port
        self.http_port = http_port
        self.metadata = metadata or []
        
        # 创建gRPC通道
        if use_tls:
            creds = grpc.ssl_channel_credentials()
            self.channel = grpc.secure_channel(
                f"{host}:{grpc_port}",
                creds,
                options=[
                    ('grpc.max_send_message_length', 50 * 1024 * 1024),
                    ('grpc.max_receive_message_length', 50 * 1024 * 1024),
                ]
            )
        else:
            self.channel = grpc.insecure_channel(
                f"{host}:{grpc_port}",
                options=[
                    ('grpc.max_send_message_length', 50 * 1024 * 1024),
                    ('grpc.max_receive_message_length', 50 * 1024 * 1024),
                ]
            )
        
        # 创建服务存根
        self.project = ProjectServiceStub(self.channel)
        self.execution = ExecutionServiceStub(self.channel)
    
    def close(self):
        """关闭通道"""
        self.channel.close()
    
    def __enter__(self):
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()


class ProjectServiceStub:
    """项目服务客户端封装"""
    
    def __init__(self, channel: grpc.Channel):
        self._stub = project_pb2_grpc.ProjectServiceStub(channel)
    
    def create_project(
        self,
        name: str,
        description: str = "",
        collaborator_ids: list = None,
        metadata: dict = None,
        timeout: float = 30.0,
        metadata_tuple: list = None
    ) -> project_pb2.Project:
        """创建项目"""
        request = project_pb2.CreateProjectRequest(
            name=name,
            description=description,
            collaborator_ids=collaborator_ids or [],
            metadata=metadata or {}
        )
        
        return self._stub.CreateProject(
            request,
            timeout=timeout,
            metadata=metadata_tuple
        )
    
    def get_project(
        self,
        project_id: str,
        timeout: float = 10.0,
        metadata_tuple: list = None
    ) -> project_pb2.Project:
        """获取项目"""
        request = project_pb2.GetProjectRequest(id=project_id)
        response = self._stub.GetProject(
            request,
            timeout=timeout,
            metadata=metadata_tuple
        )
        return response.project
    
    def list_user_projects(
        self,
        user_id: str,
        page_size: int = 20,
        page_token: str = "",
        timeout: float = 30.0,
        metadata_tuple: list = None
    ) -> tuple:
        """列出用户项目"""
        request = project_pb2.ListUserProjectsRequest(
            user_id=user_id,
            page_size=page_size,
            page_token=page_token
        )
        
        response = self._stub.ListUserProjects(
            request,
            timeout=timeout,
            metadata=metadata_tuple
        )
        
        return (
            list(response.projects),
            response.next_page_token,
            response.total_size
        )
    
    def watch_projects(
        self,
        project_ids: list,
        timeout: float = 300.0,
        metadata_tuple: list = None
    ) -> Iterator[project_pb2.WatchProjectsResponse]:
        """监听项目变更"""
        request = project_pb2.WatchProjectsRequest(project_ids=project_ids)
        
        return self._stub.WatchProjects(
            request,
            timeout=timeout,
            metadata=metadata_tuple
        )
    
    def update_project(
        self,
        project_id: str,
        name: str = None,
        description: str = None,
        status: project_pb2.ProjectStatus = None,
        metadata: dict = None,
        timeout: float = 30.0,
        metadata_tuple: list = None
    ) -> project_pb2.Project:
        """更新项目"""
        request = project_pb2.UpdateProjectRequest(
            id=project_id,
            name=name or "",
            description=description or "",
            status=status or project_pb2.ProjectStatus.ProjectStatus(0),
            metadata=metadata or {}
        )
        
        response = self._stub.UpdateProject(
            request,
            timeout=timeout,
            metadata=metadata_tuple
        )
        
        return response.project
    
    def delete_project(
        self,
        project_id: str,
        timeout: float = 30.0,
        metadata_tuple: list = None
    ) -> bool:
        """删除项目"""
        request = project_pb2.DeleteProjectRequest(id=project_id)
        response = self._stub.DeleteProject(
            request,
            timeout=timeout,
            metadata=metadata_tuple
        )
        return response.success


class ExecutionServiceStub:
    """执行服务客户端封装"""
    
    def __init__(self, channel: grpc.Channel):
        self._stub = execution_pb2_grpc.ExecutionServiceStub(channel)
    
    def execute(
        self,
        language: str,
        code: str,
        execution_id: str = None,
        timeout_ms: int = 30000,
        memory_limit_mb: int = 256,
        environment: dict = None,
        stdin_input: str = "",
        timeout: float = 60.0,
        metadata_tuple: list = None
    ) -> execution_pb2.ExecutionResponse:
        """执行代码"""
        if not execution_id:
            execution_id = f"exec-{int(time.time() * 1000)}"
        
        request = execution_pb2.ExecutionRequest(
            execution_id=execution_id,
            language=language,
            code=code,
            timeout_ms=timeout_ms,
            memory_limit_mb=memory_limit_mb,
            environment=environment or {},
            stdin_input=stdin_input
        )
        
        return self._stub.Execute(
            request,
            timeout=timeout,
            metadata=metadata_tuple
        )
    
    def execute_stream(
        self,
        language: str,
        code: str,
        execution_id: str = None,
        timeout_ms: int = 30000,
        timeout: float = 60.0,
        metadata_tuple: list = None
    ) -> Iterator[execution_pb2.ExecutionResponse]:
        """流式执行代码"""
        if not execution_id:
            execution_id = f"exec-{int(time.time() * 1000)}"
        
        request = execution_pb2.ExecutionRequest(
            execution_id=execution_id,
            language=language,
            code=code,
            timeout_ms=timeout_ms
        )
        
        return self._stub.ExecuteStream(
            request,
            timeout=timeout,
            metadata=metadata_tuple
        )


# 使用示例
if __name__ == "__main__":
    with AIIDEClient("localhost", 50051) as client:
        # 创建项目
        project = client.project.create_project(
            name="test-project",
            description="A test project",
            metadata={"env": "development"}
        )
        print(f"Created project: {project.id}, name={project.name}")
        
        # 获取项目
        fetched = client.project.get_project(project.id)
        print(f"Fetched project: {fetched.name}, status={fetched.status}")
        
        # 列出项目
        projects, next_token, total = client.project.list_user_projects("user-123")
        print(f"User has {total} projects, first page: {len(projects)}")
        
        # 执行代码
        result = client.execution.execute(
            language="python",
            code="print('Hello, World!')",
            timeout_ms=5000
        )
        print(f"Execution result: stdout={result.stdout}, exit_code={result.exit_code}")