laic-backend/service/live_callback_service.go

package service

import (
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"time"

	"gorm.io/gorm"

	"laic-backend/common"
	"laic-backend/model"
)

func (s *LiveService) HandleAliyunCallback(eventType, appName, streamName string, payload []byte) error {
	if (eventType != "publish" && eventType != "publish_done") || appName != common.AppConf.Live.AppName || streamName == "" {
		return nil
	}
	sum := sha256.Sum256(payload)
	payloadHash := hex.EncodeToString(sum[:])
	idSum := sha256.Sum256([]byte(eventType + "\n" + appName + "\n" + streamName + "\n" + payloadHash))
	event := model.LiveCallbackEvent{
		EventID:     hex.EncodeToString(idSum[:]),
		StreamName:  streamName,
		EventType:   eventType,
		PayloadHash: payloadHash,
		ReceivedAt:  time.Now(),
	}
	if err := common.DB.Create(&event).Error; err != nil {
		if code, _ := common.ParseError(err); code == 1062 || errors.Is(err, gorm.ErrDuplicatedKey) {
			return nil
		}
		return err
	}
	now := time.Now()
	updates := map[string]any{"cloud_checked_at": now}
	if eventType == "publish" {
		updates["cloud_online"] = true
	} else {
		updates["cloud_online"] = false
	}
	if err := common.DB.Model(&model.LiveSession{}).Where("stream_name = ? AND phase IN ('starting','streaming','reconnecting','stopping')", streamName).Updates(updates).Error; err != nil {
		return err
	}
	return common.DB.Model(&event).Update("processed_at", now).Error
}