bj_power/bj_power_mes/internal/eventlog/writer.go

package eventlog

import (
	"context"
	"fmt"
	"log/slog"
	"sync"
	"time"

	"bj_power_mes/ent"
	enteventlog "bj_power_mes/ent/eventlog"
	"bj_power_mes/internal/eventbus"
)

// EventLogWriter 异步批量写入事件日志到 PostgreSQL
type EventLogWriter struct {
	entClient *ent.Client
	ch        chan eventbus.Event
	batchSize int
	flushMs   int

	mu      sync.Mutex
	running bool
	cancel  context.CancelFunc
	wg      sync.WaitGroup
}

// NewEventLogWriter 创建事件日志写入器
func NewEventLogWriter(entClient *ent.Client, batchSize, flushMs int) *EventLogWriter {
	if batchSize <= 0 {
		batchSize = 20
	}
	if flushMs <= 0 {
		flushMs = 1000
	}
	return &EventLogWriter{
		entClient: entClient,
		ch:        make(chan eventbus.Event, 256),
		batchSize: batchSize,
		flushMs:   flushMs,
	}
}

// Start 启动异步写入循环
func (w *EventLogWriter) Start(ctx context.Context) {
	w.mu.Lock()
	if w.running {
		w.mu.Unlock()
		return
	}
	w.running = true
	ctx, w.cancel = context.WithCancel(ctx)
	w.mu.Unlock()

	w.wg.Add(1)
	go w.writeLoop(ctx)
}

// Stop 停止写入器
func (w *EventLogWriter) Stop() {
	w.mu.Lock()
	if !w.running {
		w.mu.Unlock()
		return
	}
	w.cancel()
	w.mu.Unlock()
	w.wg.Wait()
}

// Write 写入事件（非阻塞，满时丢弃）
func (w *EventLogWriter) Write(event eventbus.Event) {
	select {
	case w.ch <- event:
	default:
		slog.Warn("eventlog: channel full, dropping event", "type", event.Type)
	}
}

func (w *EventLogWriter) writeLoop(ctx context.Context) {
	defer w.wg.Done()

	ticker := time.NewTicker(time.Duration(w.flushMs) * time.Millisecond)
	defer ticker.Stop()

	var batch []eventbus.Event
	for {
		select {
		case <-ctx.Done():
			w.flush(batch)
			return
		case event := <-w.ch:
			batch = append(batch, event)
			if len(batch) >= w.batchSize {
				w.flush(batch)
				batch = batch[:0]
			}
		case <-ticker.C:
			if len(batch) > 0 {
				w.flush(batch)
				batch = batch[:0]
			}
		}
	}
}

func (w *EventLogWriter) flush(events []eventbus.Event) {
	if len(events) == 0 {
		return
	}
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	var builders []*ent.EventLogCreate
	for _, e := range events {
		// 连接状态事件仅用于 SSE 实时推送，不持久化
		if e.Type == eventbus.EventConnectionStatusUpdated {
			continue
		}
		// 报警确认仅用于 SSE 实时推送，不持久化
		if e.Type == eventbus.EventAlarmAcked {
			continue
		}
		// MACHINE_DONE: 创建独立事件记录（与 MACHINE_START 配对）
		// 同时更新对应 MACHINE_START 的 endedAt 用于耗时查询
		if e.Type == eventbus.EventMachineDone {
			if endedAt, ok := e.Payload["endedAt"].(time.Time); ok {
				jobId, _ := getPayloadInt(e.Payload, "jobId")
				equipId, _ := getPayloadInt(e.Payload, "equipmentId")
				if jobId > 0 && equipId > 0 {
					w.updateMachineStartEndedAt(ctx, jobId, equipId, endedAt)
				}
			}
		}
		// INSPECTION_RESULT: 同样更新对应 MACHINE_START 的 endedAt
		if e.Type == eventbus.EventInspectionResult {
			if endedAt, ok := e.Payload["endedAt"].(time.Time); ok {
				jobId, _ := getPayloadInt(e.Payload, "jobId")
				equipId, _ := getPayloadInt(e.Payload, "equipmentId")
				if jobId > 0 && equipId > 0 {
					w.updateMachineStartEndedAt(ctx, jobId, equipId, endedAt)
				}
			}
		}

		b := w.entClient.EventLog.Create().
			SetEventType(string(e.Type)).
			SetSourceId(e.Source).
			SetEntityType(e.EntityType).
			SetEntityId(e.EntityID).
			SetEntityVersion(e.EntityVersion).
			SetDescription(eventDescription(e)).
			SetPayload(e.Payload)
		if v, ok := getPayloadInt(e.Payload, "jobId"); ok {
			b.SetJobId(v)
		}
		if v, ok := e.Payload["workpieceNo"].(string); ok && v != "" {
			b.SetWorkpieceNo(v)
		}
		if v, ok := e.Payload["sn"].(string); ok && v != "" && e.Payload["workpieceNo"] == nil {
			b.SetWorkpieceNo(v) // 工位终端上报以 sn 为工件号
		}
		if v, ok := e.Payload["workOrderNo"].(string); ok && v != "" {
			b.SetWorkOrderNo(v)
		} else if v, ok := e.Payload["orderNo"].(string); ok && v != "" {
			b.SetWorkOrderNo(v)
		}
		if v, ok := e.Payload["operator"].(string); ok && v != "" {
			b.SetOperator(v)
		}
		if v, ok := getPayloadInt(e.Payload, "equipmentId"); ok {
			b.SetEquipmentId(v)
		}
		if v, ok := e.Payload["equipmentName"].(string); ok && v != "" {
			b.SetEquipmentName(v)
		}
		if v, ok := getPayloadInt(e.Payload, "tempSlotNo"); ok {
			b.SetTempSlotNo(v)
		}
		if v, ok := e.Payload["startedAt"].(time.Time); ok {
			b.SetStartedAt(v)
		}
		if v, ok := e.Payload["endedAt"].(time.Time); ok {
			b.SetEndedAt(v)
		}
		builders = append(builders, b)
	}
	if len(builders) > 0 {
		if _, err := w.entClient.EventLog.CreateBulk(builders...).Save(ctx); err != nil {
			slog.Error("eventlog: bulk save failed", "count", len(builders), "error", err)
		}
	}
}

// updateMachineStartEndedAt 更新对应 MACHINE_START 事件日志的 endedAt
func (w *EventLogWriter) updateMachineStartEndedAt(ctx context.Context, jobId, equipId int, endedAt time.Time) {
	n, err := w.entClient.EventLog.Update().
		Where(
			enteventlog.EventTypeEQ(string(eventbus.EventMachineStart)),
			enteventlog.JobIdEQ(jobId),
			enteventlog.EquipmentIdEQ(equipId),
			enteventlog.EndedAtIsNil(),
		).
		SetEndedAt(endedAt).
		Save(ctx)
	if err != nil {
		slog.Error("eventlog: update MACHINE_START endedAt failed",
			"jobId", jobId, "equipmentId", equipId, "error", err)
		return
	}
	if n == 0 {
		slog.Warn("eventlog: no matching MACHINE_START found for endedAt update",
			"jobId", jobId, "equipmentId", equipId)
	}
}

// eventDescription 根据事件类型生成中文描述
// EventDescription 根据事件类型生成中文描述（供 SSE 桥接等复用）
func EventDescription(e eventbus.Event) string {
	return eventDescription(e)
}

func eventDescription(e eventbus.Event) string {
	switch e.Type {
	case eventbus.EventLoad:
		return fmt.Sprintf("%s 从 暂存台#%s 上料至 %s", jobLabel(e), tempSlotLabel(e), equipLabel(e))
	case eventbus.EventUnload:
		return fmt.Sprintf("%s 从 %s 下料至 暂存台#%s", jobLabel(e), equipLabel(e), tempSlotLabel(e))
	case eventbus.EventExchange:
		return fmt.Sprintf("%s 在 %s 换料", jobLabel(e), equipLabel(e))
	case eventbus.EventScan:
		return jobLabel(e) + " 扫码"
	case eventbus.EventMark:
		return jobLabel(e) + " 打标"
	case eventbus.EventReplenish:
		return fmt.Sprintf("%s 从 %s 补料至 暂存台#%s", jobLabel(e), dockLabel(e), tempSlotLabel(e))
	case eventbus.EventMachineStart:
		return fmt.Sprintf("%s 在 %s 加工", jobLabel(e), equipLabel(e))
	case eventbus.EventMachineDone:
		desc := fmt.Sprintf("%s 在 %s 加工完成", jobLabel(e), equipLabel(e))
		if dur, ok := e.Payload["duration"].(string); ok && dur != "" {
			desc = fmt.Sprintf("%s 在 %s 加工完成（耗时%s）", jobLabel(e), equipLabel(e), dur)
		}
		return desc
	case eventbus.EventInspectionResult:
		if pass, ok := e.Payload["pass"].(bool); ok {
			if pass {
				return fmt.Sprintf("%s 在 %s 检测合格", jobLabel(e), equipLabel(e))
			}
			return fmt.Sprintf("%s 在 %s 检测不合格", jobLabel(e), equipLabel(e))
		}
		return fmt.Sprintf("%s 在 %s 检测", jobLabel(e), equipLabel(e))
	case eventbus.EventJobCompleted:
		return jobLabel(e) + " 全部工序完成"
	case eventbus.EventJobScrapped:
		return jobLabel(e) + " 报废"
	case eventbus.EventJobDischarged:
		return fmt.Sprintf("%s 从 暂存台#%s 下线至 %s", jobLabel(e), tempSlotLabel(e), dockLabel(e))

	case eventbus.EventOrderStarted:
		return fmt.Sprintf("%s 启动", orderLabel(e))
	case eventbus.EventOrderPaused:
		return fmt.Sprintf("%s 暂停", orderLabel(e))
	case eventbus.EventOrderResumed:
		return fmt.Sprintf("%s 恢复", orderLabel(e))
	case eventbus.EventOrderCancelled:
		return fmt.Sprintf("%s 取消", orderLabel(e))
	case eventbus.EventOrderCompleted:
		return fmt.Sprintf("%s 完成", orderLabel(e))

	case eventbus.EventPalletCompleted:
		dockNo, _ := getPayloadInt(e.Payload, "dockNo")
		passQty, _ := getPayloadInt(e.Payload, "passQty")
		ngQty, _ := getPayloadInt(e.Payload, "ngQty")
		return fmt.Sprintf("接驳台#%d 托盘完工（良品%d 不良%d）", dockNo, passQty, ngQty)

	case eventbus.EventAlarmRaised:
		alarmCode, _ := e.Payload["alarmCode"].(string)
		alarmMsg, _ := e.Payload["alarmMessage"].(string)
		entityLabel := entityLabel(e.EntityType, e.EntityID)
		return fmt.Sprintf("%s 报警: [%s] %s", entityLabel, alarmCode, alarmMsg)

	case eventbus.EventAlarmAcked:
		entityLabel := entityLabel(e.EntityType, e.EntityID)
		return fmt.Sprintf("%s 报警已确认", entityLabel)

	case eventbus.EventEquipmentAvailabilityChanged:
		enabled, _ := e.Payload["enabled"].(bool)
		if enabled {
			return fmt.Sprintf("%s 已设为调度可用", equipLabel(e))
		}
		return fmt.Sprintf("%s 已设为调度不可用", equipLabel(e))

	case eventbus.EventManualActionCreated:
		actionType, _ := e.Payload["actionType"].(string)
		entityLabel := entityLabel(e.EntityType, e.EntityID)
		return fmt.Sprintf("%s 创建人工处理: %s", entityLabel, manualActionTypeDesc(actionType))

	case eventbus.EventManualActionResolved:
		entityLabel := entityLabel(e.EntityType, e.EntityID)
		return fmt.Sprintf("%s 人工处理已解决", entityLabel)

	default:
		return fmt.Sprintf("%s: %s", entityLabel(e.EntityType, e.EntityID), string(e.Type))
	}
}

// jobLabel 生成 "工件 #344 (M2026...)" 格式
func jobLabel(e eventbus.Event) string {
	if v, ok := getPayloadInt(e.Payload, "jobId"); ok && v > 0 {
		if wn, ok := e.Payload["workpieceNo"].(string); ok && wn != "" {
			return fmt.Sprintf("工件 #%d (%s)", v, wn)
		}
		return fmt.Sprintf("工件 #%d", v)
	}
	return fmt.Sprintf("%s #%s", entityLabel(e.EntityType, e.EntityID), e.EntityID)
}

// equipLabel 生成 "CNC#3" 格式
func equipLabel(e eventbus.Event) string {
	en, _ := e.Payload["equipmentName"].(string)
	if v, ok := getPayloadInt(e.Payload, "equipmentId"); ok && v > 0 {
		if en != "" {
			return fmt.Sprintf("%s#%d", en, v)
		}
		return fmt.Sprintf("设备#%d", v)
	}
	if en != "" {
		return en
	}
	return "设备"
}

// tempSlotLabel 生成 "7" 格式（纯槽位号）
func tempSlotLabel(e eventbus.Event) string {
	if v, ok := getPayloadInt(e.Payload, "tempSlotNo"); ok && v > 0 {
		return fmt.Sprintf("%d", v)
	}
	if v, ok := getPayloadInt(e.Payload, "slotNo"); ok && v > 0 {
		return fmt.Sprintf("%d", v)
	}
	return ""
}

func getPayloadInt(payload map[string]any, key string) (int, bool) {
	if v, ok := payload[key].(float64); ok {
		return int(v), true
	}
	if v, ok := payload[key].(int); ok {
		return v, true
	}
	return 0, false
}

// dockLabel 生成 接驳台1-3 格式
func dockLabel(e eventbus.Event) string {
	dn, dok := getPayloadInt(e.Payload, "dockNo")
	ds, sok := getPayloadInt(e.Payload, "dockSlotNo")
	if dok && dn > 0 && sok && ds > 0 {
		return fmt.Sprintf("接驳台#%d-%d", dn, ds)
	}
	return ""
}

// orderLabel 生成 "工单 #42" 格式
func orderLabel(e eventbus.Event) string {
	if v, ok := getPayloadInt(e.Payload, "orderId"); ok && v > 0 {
		return fmt.Sprintf("工单 #%d", v)
	}
	return fmt.Sprintf("%s #%s", entityLabel(e.EntityType, e.EntityID), e.EntityID)
}

func entityLabel(entityType, entityID string) string {
	switch entityType {
	case "job":
		return fmt.Sprintf("工件 #%s", entityID)
	case "workorder":
		return fmt.Sprintf("工单 #%s", entityID)
	case "equipment":
		return fmt.Sprintf("设备 #%s", entityID)
	default:
		if entityType != "" && entityID != "" {
			return fmt.Sprintf("%s #%s", entityType, entityID)
		}
		return "系统"
	}
}

func manualActionTypeDesc(s string) string {
	switch s {
	case "SCAN_FAILED":
		return "扫码失败"
	case "RECOVERY_REQUIRED":
		return "需恢复确认"
	case "INSPECTION_DECISION":
		return "检测判定"
	case "FAULT_CONFIRM":
		return "故障确认"
	default:
		return s
	}
}