三、服务注册与发现

// 写死服务端地址
client, err := NewClientV2("127.0.0.1:9091", &JSONSerializer{})

package registry

import (
	"context"
	"time"
)

// ============================================================
// 第一部分：核心数据结构与接口定义
// ============================================================

// ServiceInstance 代表一个服务实例的信息
// 每个运行中的服务进程对应一个 ServiceInstance
type ServiceInstance struct {
	ServiceName string            `json:"serviceName"` // 服务名，例如 "user-service"
	InstanceID  string            `json:"instanceId"`  // 实例ID，唯一标识一个实例，例如 "172.10.0.1:9091"
	Address     string            `json:"address"`     // 服务地址（IP），例如 "172.10.0.1"
	Port        int               `json:"port"`        // 服务端口，例如 9091
	Metadata    map[string]string `json:"metadata"`    // 元数据，例如版本号、权重等
}

// Registry 是注册中心的抽象接口
// 不同的注册中心实现（etcd、consul、zookeeper）都实现这个接口
// 接口方法分成三部分：服务注册、服务发现、监听服务
type Registry interface {
	// --- 服务注册部分 ---

	// Register 将服务实例注册到注册中心
	// 参数 ctx 用于控制超时和取消
	// 参数 instance 是要注册的服务实例信息
	// 返回的 cancel 函数用于取消注册（服务下线时调用）
	Register(ctx context.Context, instance *ServiceInstance) (cancel context.CancelFunc, err error)

	// Unregister 从注册中心取消注册
	// 服务正常退出时调用，主动删除注册数据
	Unregister(ctx context.Context, instance *ServiceInstance) error

	// --- 服务发现部分 ---

	// ListServices 根据服务名列出所有可用的服务实例
	// 客户端启动时调用，拉取当前可用的服务列表
	ListServices(ctx context.Context, serviceName string) ([]*ServiceInstance, error)

	// --- 监听服务部分 ---

	// Subscribe 订阅服务变更
	// 当服务实例上线/下线时，通过 channel 通知调用方
	// 返回的 channel 会接收服务实例的变更事件
	Subscribe(ctx context.Context, serviceName string) (<-chan Event, error)
}

// Event 代表服务变更事件
type Event struct {
	Type     EventType         // 事件类型：新增或删除
	Instance *ServiceInstance  // 变更的服务实例
}

// EventType 事件类型
type EventType uint8

const (
	EventTypeAdd    EventType = iota // 服务实例新增
	EventTypeDelete                  // 服务实例删除
)

package registry

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

	clientv3 "go.etcd.io/etcd/client/v3"
)

// ============================================================
// 第二部分：etcd 注册中心实现
// ============================================================

// EtcdRegistry 基于 etcd 的注册中心实现
// 利用 etcd 的 Lease（租约）机制实现服务健康检查
// 利用 etcd 的 Watch（监听）机制实现服务变更通知
type EtcdRegistry struct {
	client    *clientv3.Client  // etcd 客户端连接
	keyPrefix string            // etcd 中 key 的前缀，例如 "/micro/"

	// 租约相关配置
	leaseTTL      int64         // 租约存活时间（秒），例如 15 秒
	keepaliveOnce sync.Once     // 确保心跳 goroutine 只启动一次

	// 每个服务实例对应的租约 ID
	// key: instance key, value: lease ID
	leases sync.Map             // 并发安全的 map，存储实例与租约的映射
}

// NewEtcdRegistry 创建一个新的 etcd 注册中心
// 参数 endpoints 是 etcd 集群的地址列表，例如 []string{"127.0.0.1:2379"}
func NewEtcdRegistry(endpoints []string) (*EtcdRegistry, error) {
	// 创建 etcd 客户端
	client, err := clientv3.New(clientv3.Config{
		Endpoints:   endpoints,              // etcd 地址列表
		DialTimeout: 5 * time.Second,       // 连接超时
	})
	if err != nil {
		return nil, fmt.Errorf("创建 etcd 客户端失败: %w", err)
	}

	return &EtcdRegistry{
		client:    client,
		keyPrefix: "/micro/",  // key 前缀，所有服务注册数据都存放在这个前缀下
		leaseTTL:  15,         // 租约 15 秒，即服务 15 秒不续约就认为宕机
	}, nil
}

// instanceKey 构造实例在 etcd 中的 key
// 格式：/micro/service-name/instance-id
func (r *EtcdRegistry) instanceKey(inst *ServiceInstance) string {
	return fmt.Sprintf("%s%s/%s", r.keyPrefix, inst.ServiceName, inst.InstanceID)
}

// servicePrefix 构造服务在 etcd 中的前缀
// 格式：/micro/service-name/
// 用于 ListServices 时按前缀查询所有实例
func (r *EtcdRegistry) servicePrefix(serviceName string) string {
	return fmt.Sprintf("%s%s/", r.keyPrefix, serviceName)
}

// ============================================================
// 第三部分：服务注册实现
// ============================================================

// Register 将服务实例注册到 etcd
// 核心步骤：
// 1. 创建租约（Lease）
// 2. 将实例信息以 JSON 格式写入 etcd，绑定租约
// 3. 启动 goroutine 持续续约（KeepAlive）
func (r *EtcdRegistry) Register(ctx context.Context, instance *ServiceInstance) (context.CancelFunc, error) {
	// 第一步：创建租约
	// 租约是一个存活时间，超过这个时间没有续约，数据就会被自动删除
	// 这就是服务宕机后能自动从注册中心移除的关键
	leaseResp, err := r.client.Grant(ctx, r.leaseTTL)
	if err != nil {
		return nil, fmt.Errorf("创建租约失败: %w", err)
	}
	leaseID := leaseResp.ID

	// 第二步：将实例信息序列化为 JSON
	data, err := json.Marshal(instance)
	if err != nil {
		return nil, fmt.Errorf("序列化实例信息失败: %w", err)
	}

	// 第三步：将数据写入 etcd，绑定租约
	// key = /micro/service-name/instance-id
	// value = JSON 格式的实例信息
	// lease = 刚创建的租约 ID
	key := r.instanceKey(instance)
	_, err = r.client.Put(ctx, key, string(data), clientv3.WithLease(leaseID))
	if err != nil {
		return nil, fmt.Errorf("写入 etcd 失败: %w", err)
	}

	// 记录租约 ID，用于后续取消注册
	r.leases.Store(key, leaseID)

	// 第四步：启动续约 goroutine
	// KeepAlive 会持续向 etcd 发送续约请求，保持租约不过期
	// 如果服务宕机，这个 goroutine 也会停止，租约就会过期
	keepAliveCtx, cancel := context.WithCancel(context.Background())
	go r.keepAlive(keepAliveCtx, leaseID)

	// 返回 cancel 函数，调用方可以在服务下线时调用
	// 调用 cancel 会停止续约 goroutine，租约随后过期，数据被删除
	return cancel, nil
}

// keepAlive 持续向 etcd 续约
// 只要这个 goroutine 在运行，租约就不会过期
// 当 ctx 被取消时（服务下线），续约停止，租约过期
func (r *EtcdRegistry) keepAlive(ctx context.Context, leaseID clientv3.LeaseID) {
	// client.KeepAlive 返回一个 channel
	// 每次续约成功，channel 就会收到一个响应
	// 如果 ctx 取消或 etcd 连接断开，channel 会被关闭
	ch, err := r.client.KeepAlive(ctx, leaseID)
	if err != nil {
		// 续约失败，租约会在 leaseTTL 秒后过期
		return
	}

	// 持续读取续约响应
	// 如果 ctx 被取消，KeepAlive 会关闭 channel，循环退出
	for range ch {
		// 收到续约响应，什么都不用做
		// 只要从 channel 能读到数据，就说明续约成功
	}
}

// Unregister 从 etcd 取消注册
// 服务正常退出时调用，主动删除注册数据
func (r *EtcdRegistry) Unregister(ctx context.Context, instance *ServiceInstance) error {
	key := r.instanceKey(instance)

	// 从 etcd 中删除数据
	_, err := r.client.Delete(ctx, key)
	if err != nil {
		return fmt.Errorf("删除注册数据失败: %w", err)
	}

	// 清理租约记录
	r.leases.Delete(key)
	return nil
}

// 服务注册的使用示例
func main() {
	// 创建 etcd 注册中心
	registry, err := NewEtcdRegistry([]string{"127.0.0.1:2379"})
	if err != nil {
		panic(err)
	}

	// 创建服务实例信息
	instance := &ServiceInstance{
		ServiceName: "user-service",
		InstanceID:  "172.10.0.1:9091", // 用 IP:Port 作为实例ID
		Address:     "172.10.0.1",
		Port:        9091,
		Metadata: map[string]string{
			"version": "v1.0.0",
			"weight":  "100",
		},
	}

	// 注意：用户必须确保所有初始化工作都完成后才调用注册
	// 例如数据库连接池已创建、缓存已预热等
	ctx := context.Background()
	cancel, err := registry.Register(ctx, instance)
	if err != nil {
		panic(err)
	}

	// 启动 gRPC 服务（省略具体代码）
	// servegrpc()

	// 服务退出时取消注册
	defer func() {
		cancel()                   // 停止续约
		registry.Unregister(ctx, instance) // 主动删除注册数据
	}()
}

package registry

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

	"google.golang.org/grpc/resolver"
	clientv3 "go.etcd.io/etcd/client/v3"
)

// ============================================================
// 第四部分：gRPC Resolver 和 Builder 实现
// ============================================================

// etcdBuilder 实现 resolver.Builder 接口
// 负责根据 scheme 创建 Resolver
type etcdBuilder struct {
	client    *clientv3.Client // etcd 客户端
	keyPrefix string           // key 前缀
}

// NewEtcdBuilder 创建一个 etcd resolver builder
// 参数 scheme 是注册中心的协议标识，例如 "etcd"
// 这样 gRPC 拨号时使用 "etcd:///service-name" 就会使用这个 builder
func NewEtcdBuilder(client *clientv3.Client, scheme, keyPrefix string) *etcdBuilder {
	// 将 builder 注册到 gRPC 的 scheme 映射表中
	// gRPC 内部维护了一个 scheme -> builder 的 map
	resolver.Register(&etcdBuilder{
		client:    client,
		keyPrefix: keyPrefix,
	})

	return &etcdBuilder{
		client:    client,
		keyPrefix: keyPrefix,
	}
}

// Build 是 resolver.Builder 接口的核心方法
// 当 gRPC 拨号时遇到对应的 scheme，就会调用这个方法创建 Resolver
// 参数：
//   - target: 拨号目标，包含 scheme 和服务名
//   - cc: ClientConn，用于更新可用地址列表
//   - opts: resolver 选项
func (b *etcdBuilder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
	// 创建 etcd Resolver
	r := &etcdResolver{
		client:    b.client,
		keyPrefix: b.keyPrefix,
		target:    target,           // 拨号目标（包含服务名）
		cc:        cc,               // ClientConn，用于更新地址
	}

	// 启动一个 goroutine 来处理服务发现
	go r.start()

	return r, nil
}

// Scheme 返回这个 builder 对应的 scheme
// gRPC 通过这个方法来匹配拨号地址中的 scheme
func (b *etcdBuilder) Scheme() string {
	return "etcd" // 对应 "etcd:///service-name" 中的 "etcd"
}

// ============================================================
// 第五部分：etcd Resolver 实现
// ============================================================

// etcdResolver 实现 resolver.Resolver 接口
// 负责从 etcd 拉取服务列表，并监听变更
type etcdResolver struct {
	client    *clientv3.Client  // etcd 客户端
	keyPrefix string            // key 前缀
	target    resolver.Target   // 拨号目标
	cc        resolver.ClientConn // gRPC 连接，用于更新地址列表
	mu        sync.Mutex        // 保护 addrs 的并发访问
}

// start 启动服务发现
// 1. 先从 etcd 拉取当前所有服务实例
// 2. 然后监听 etcd 中服务信息的变更
func (r *etcdResolver) start() {
	serviceName := r.target.Endpoint() // 从 target 中提取服务名
	prefix := fmt.Sprintf("%s%s/", r.keyPrefix, serviceName)

	// 第一步：从 etcd 拉取所有服务实例
	// 使用 Get + WithPrefix 查询指定前缀下的所有 key
	resp, err := r.client.Get(context.Background(), prefix, clientv3.WithPrefix())
	if err != nil {
		return
	}

	// 将拉取到的实例信息转换为 gRPC 的地址列表
	var addrs []resolver.Address
	for _, kv := range resp.Kvs {
		var instance ServiceInstance
		if err := json.Unmarshal(kv.Value, &instance); err != nil {
			continue // 跳过无法解析的数据
		}
		// 构造 gRPC 地址
		addrs = append(addrs, resolver.Address{
			Addr:       fmt.Sprintf("%s:%d", instance.Address, instance.Port),
			ServerName: instance.ServiceName,
		})
	}

	// 更新 gRPC 的连接地址列表
	// gRPC 内部会根据这个列表创建/维护 TCP 连接
	r.cc.UpdateState(resolver.State{
		Addresses: addrs,
	})

	// 第二步：监听 etcd 中服务信息的变更
	// 使用 Watch + WithPrefix 监听指定前缀下所有 key 的变更
	// 当有服务实例上线/下线时，etcd 会通过 channel 发送事件
	go r.watch(prefix)
}

// watch 监听 etcd 中服务信息的变更
// 当服务实例上线或下线时，更新 gRPC 的地址列表
func (r *etcdResolver) watch(prefix string) {
	// 创建 Watch channel
	// WithPrefix 表示监听指定前缀下所有 key 的变更
	// WithPrevKV 表示删除事件中会包含被删除 key 的旧值
	watchCh := r.client.Watch(context.Background(), prefix,
		clientv3.WithPrefix(), clientv3.WithPrevKV())

	for watchResp := range watchCh {
		for _, event := range watchResp.Events {
			switch event.Type {
			case clientv3.EventTypePut:
				// 有新的服务实例注册（或更新）
				// 解析新实例信息
				var instance ServiceInstance
				if err := json.Unmarshal(event.Kv.Value, &instance); err != nil {
					continue
				}
				// 这里采用简单策略：收到任何变更就重新拉取全部数据
				// 更精细的做法是只增加/删除变更的实例
				r.refreshAddresses(prefix)

			case clientv3.EventTypeDelete:
				// 有服务实例下线
				// event.PrevKv 包含被删除的旧值
				r.refreshAddresses(prefix)
			}
		}
	}
}

// refreshAddresses 重新从 etcd 拉取所有服务实例，更新地址列表
// 这对应 PDF 中提到的第一种实现思路：
// "监听到事件之后，不管事件的类型，直接拉取所有的服务信息，完整更新一下"
func (r *etcdResolver) refreshAddresses(prefix string) {
	resp, err := r.client.Get(context.Background(), prefix, clientv3.WithPrefix())
	if err != nil {
		return
	}

	var addrs []resolver.Address
	for _, kv := range resp.Kvs {
		var instance ServiceInstance
		if err := json.Unmarshal(kv.Value, &instance); err != nil {
			continue
		}
		addrs = append(addrs, resolver.Address{
			Addr:       fmt.Sprintf("%s:%d", instance.Address, instance.Port),
			ServerName: instance.ServiceName,
		})
	}

	r.cc.UpdateState(resolver.State{
		Addresses: addrs,
	})
}

// ResolveNow resolver.Resolver 接口要求实现的方法
// gRPC 在某些情况下会调用这个方法要求立即重新解析
// 我们使用 Watch 机制实时监听，所以这里不需要做任何事
func (r *etcdResolver) ResolveNow(o resolver.ResolveNowOptions) {}

// Close 关闭 Resolver，释放资源
func (r *etcdResolver) Close() {}

// ListServices 根据服务名列出所有可用服务实例
// 客户端启动时调用，拉取当前可用的服务列表
func (r *EtcdRegistry) ListServices(ctx context.Context, serviceName string) ([]*ServiceInstance, error) {
	// 构造查询前缀：/micro/service-name/
	prefix := r.servicePrefix(serviceName)

	// 从 etcd 中查询该前缀下的所有 key
	// WithPrefix 表示查询所有以 prefix 开头的 key
	resp, err := r.client.Get(ctx, prefix, clientv3.WithPrefix())
	if err != nil {
		return nil, fmt.Errorf("查询 etcd 失败: %w", err)
	}

	// 解析查询结果
	var instances []*ServiceInstance
	for _, kv := range resp.Kvs {
		var inst ServiceInstance
		if err := json.Unmarshal(kv.Value, &inst); err != nil {
			continue // 跳过无法解析的数据
		}
		instances = append(instances, &inst)
	}

	return instances, nil
}

// Subscribe 订阅服务变更
// 通过 etcd 的 Watch 机制监听服务实例的上线/下线
func (r *EtcdRegistry) Subscribe(ctx context.Context, serviceName string) (<-chan Event, error) {
	prefix := r.servicePrefix(serviceName)

	// 创建事件 channel
	eventCh := make(chan Event, 10) // 带缓冲，防止事件丢失

	// 启动 goroutine 监听 etcd 变更
	go func() {
		defer close(eventCh)

		// 创建 Watch
		watchCh := r.client.Watch(ctx, prefix,
			clientv3.WithPrefix(), clientv3.WithPrevKV())

		for watchResp := range watchCh {
			for _, event := range watchResp.Events {
				switch event.Type {
				case clientv3.EventTypePut:
					// 新服务实例注册
					var inst ServiceInstance
					if err := json.Unmarshal(event.Kv.Value, &inst); err != nil {
						continue
					}
					eventCh <- Event{
						Type:     EventTypeAdd,
						Instance: &inst,
					}

				case clientv3.EventTypeDelete:
					// 服务实例下线
					// event.PrevKv 包含被删除的旧值
					if event.PrevKv == nil {
						continue
					}
					var inst ServiceInstance
					if err := json.Unmarshal(event.PrevKv.Value, &inst); err != nil {
						continue
					}
					eventCh <- Event{
						Type:     EventTypeDelete,
						Instance: &inst,
					}
				}
			}
		}
	}()

	return eventCh, nil
}

package main

import (
	"context"
	"fmt"
	"time"

	"google.golang.org/grpc"
	clientv3 "go.etcd.io/etcd/client/v3"
)

func main() {
	// ========================================
	// 服务端：注册服务到 etcd
	// ========================================

	// 创建 etcd 客户端
	etcdClient, err := clientv3.New(clientv3.Config{
		Endpoints:   []string{"127.0.0.1:2379"},
		DialTimeout: 5 * time.Second,
	})
	if err != nil {
		panic(err)
	}

	// 创建注册中心
	registry, err := NewEtcdRegistry([]string{"127.0.0.1:2379"})
	if err != nil {
		panic(err)
	}

	// 注册服务实例
	instance := &ServiceInstance{
		ServiceName: "user-service",
		InstanceID:  "172.10.0.1:9091",
		Address:     "172.10.0.1",
		Port:        9091,
		Metadata: map[string]string{
			"version": "v1.0.0",
		},
	}

	ctx := context.Background()
	cancel, err := registry.Register(ctx, instance)
	if err != nil {
		panic(err)
	}
	defer func() {
		cancel() // 停止续约
		registry.Unregister(ctx, instance) // 主动删除注册数据
	}()

	// 启动 gRPC 服务（省略具体代码）
	// grpcServer.Serve(lis)

	// ========================================
	// 客户端：通过 gRPC 服务发现连接服务端
	// ========================================

	// 注册 etcd resolver builder 到 gRPC
	// 这样 gRPC 拨号时遇到 "etcd" scheme 就会使用我们的 resolver
	NewEtcdBuilder(etcdClient, "etcd", "/micro/")

	// 使用服务名拨号，而不是写死 IP
	// "etcd:///user-service" 的含义：
	// - etcd: 使用 etcd resolver
	// - user-service: 要发现的服务名
	conn, err := grpc.Dial(
		"etcd:///user-service",           // 服务标识符，不写死 IP
		grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy":"round_robin"}`), // 负载均衡策略
		grpc.WithInsecure(),               // 不使用 TLS（仅用于演示）
	)
	if err != nil {
		panic(err)
	}
	defer conn.Close()

	// 此时 conn 已经自动发现了 user-service 的所有实例
	// 后续的 RPC 调用会自动负载均衡到不同的实例
	fmt.Println("已连接到 user-service")

	// 订阅服务变更（可选）
	eventCh, _ := registry.Subscribe(ctx, "user-service")
	go func() {
		for event := range eventCh {
			switch event.Type {
			case EventTypeAdd:
				fmt.Printf("新实例上线: %s:%d\n", event.Instance.Address, event.Instance.Port)
			case EventTypeDelete:
				fmt.Printf("实例下线: %s:%d\n", event.Instance.Address, event.Instance.Port)
			}
		}
	}()
}