story/internal/devicefleet/ota.go

package devicefleet

import (
	"context"
	"encoding/json"
	"errors"
	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"time"
)

// Start authorizes under the caller's device-twin lock, shared with binding changes.
func (s *Service) Start(ctx context.Context, request string, code int64, authorize func(pgx.Tx) (string, error)) (Task, error) {
	var result Task
	if s.Publish == nil {
		return result, conflict("MQTT_UNAVAILABLE", "MQTT服务未启用")
	}
	if _, err := uuid.Parse(request); err != nil {
		return result, conflict("REQUEST_ID_REQUIRED", "请求号必须为UUID")
	}
	tx, err := s.Pool.Begin(ctx)
	if err != nil {
		return result, err
	}
	defer tx.Rollback(ctx)
	device, err := authorize(tx)
	if err != nil {
		return result, err
	}
	err = tx.QueryRow(ctx, `SELECT o.id::text,o.status,f.version,f.version_code,o.message,o.created_at FROM fleet_ota o JOIN fleet_firmware f ON f.id=o.release_id WHERE o.request_id=$1 AND o.device_id=$2`, request, device).Scan(&result.ID, &result.Status, &result.Version, &result.Code, &result.Message, &result.Created)
	if err == nil {
		if result.Code != code {
			return result, conflict("REQUEST_CONFLICT", "同一请求号不能更换目标版本")
		}
		return result, nil
	}
	if !errors.Is(err, pgx.ErrNoRows) {
		return result, err
	}
	var tenant, model, factory string
	var online, supported, busy bool
	var current int64
	err = tx.QueryRow(ctx, `SELECT d.tenant_id::text,d.model,d.factory_device_id,COALESCE(p.connected AND p.last_seen_at>now()-interval '90 seconds',false),COALESCE(p.ota_supported,false),COALESCE(p.version_code,0),EXISTS(SELECT 1 FROM call_session WHERE device_id=d.id AND call_state IN ('PREPARING','STARTING','TALKING','STOPPING')) OR EXISTS(SELECT 1 FROM fleet_ota WHERE device_id=d.id AND status IN ('pending','accepted')) FROM device d LEFT JOIN fleet_presence p ON p.device_id=d.id WHERE d.id=$1 AND d.status='registered' AND d.activation_state NOT IN ('LOST','REVOKED')`, device).Scan(&tenant, &model, &factory, &online, &supported, &current, &busy)
	if err != nil {
		return result, err
	}
	if !online {
		return result, conflict("DEVICE_OFFLINE", "设备离线，请开机联网后重试")
	}
	var otherBusy bool
	err = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM conversation WHERE device_id=$1 AND status NOT IN ('COMPLETED','FAILED','EXPIRED')) OR EXISTS(SELECT 1 FROM device_command WHERE device_id=$1 AND status IN ('QUEUED','DELIVERED','ACKNOWLEDGED') AND expires_at>now()) OR EXISTS(SELECT 1 FROM firmware_assignment WHERE device_id=$1 AND status IN ('ASSIGNED','DOWNLOADING','INSTALLING'))`, device).Scan(&otherBusy)
	if err != nil {
		return result, err
	}
	busy = busy || otherBusy
	if !supported {
		return result, conflict("OTA_UNSUPPORTED", "当前设备固件或分区不支持OTA，请先由嵌入式端适配")
	}
	if busy {
		return result, conflict("DEVICE_BUSY", "设备正在通话或升级，请稍后重试")
	}
	release, err := scanRelease(tx.QueryRow(ctx, `SELECT `+releaseFields+` FROM fleet_firmware WHERE tenant_id=$1 AND model=$2 AND status='READY' ORDER BY version_code DESC LIMIT 1`, tenant, model))
	if err != nil {
		return result, err
	}
	if release.Code != code || code <= current {
		return result, conflict("VERSION_CHANGED", "版本已变化或已经是最新版本，请刷新")
	}
	_, tokenHash, downloadURL, downloadExpiresAt, err := newDownloadTicket(s.DownloadBaseURL)
	if err != nil {
		return result, conflict("DOWNLOAD_UNAVAILABLE", "固件短下载地址未正确配置")
	}
	result = Task{ID: uuid.NewString(), Status: "pending", Version: release.Version, Code: code, Created: time.Now().UTC()}
	_, err = tx.Exec(ctx, `INSERT INTO fleet_ota(id,device_id,release_id,request_id) VALUES($1,$2,$3,$4)`, result.ID, device, release.ID, request)
	if err != nil {
		return result, err
	}
	_, err = tx.Exec(ctx, `INSERT INTO fleet_firmware_download_ticket(token_hash,release_id,ota_id,expires_at) VALUES($1,$2,$3,$4)`, tokenHash, release.ID, result.ID, downloadExpiresAt)
	if err != nil {
		return result, err
	}
	if err = tx.Commit(ctx); err != nil {
		return result, err
	}
	command, _ := json.Marshal(map[string]any{"task_id": result.ID, "model": model, "version": release.Version, "version_code": code, "download_url": downloadURL, "sha256": release.SHA256, "size": release.Size, "expires_at_ms": result.Created.Add(30 * time.Second).UnixMilli()})
	// No retained commands: a device reconnecting later must not begin an obsolete OTA.
	if err = s.Publish(OTATopic(factory), command); err != nil {
		_, _ = s.Pool.Exec(ctx, `UPDATE fleet_ota SET status='failed',message='升级指令发送失败',updated_at=now() WHERE id=$1 AND status='pending'`, result.ID)
		_, _ = s.Pool.Exec(ctx, `UPDATE fleet_firmware_download_ticket SET revoked_at=now() WHERE ota_id=$1 AND revoked_at IS NULL`, result.ID)
		return result, conflict("MQTT_PUBLISH_FAILED", "升级指令发送失败，请刷新后重试")
	}
	return result, nil
}
func (s *Service) Expire(ctx context.Context) error {
	tx, err := s.Pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer tx.Rollback(ctx)
	_, err = tx.Exec(ctx, `UPDATE fleet_ota SET status='timeout',message=CASE WHEN status='pending' THEN '设备未确认接收指令' ELSE '升级结果超时，请检查设备' END,updated_at=now() WHERE (status='pending' AND created_at<now()-interval '30 seconds') OR (status='accepted' AND created_at<now()-interval '20 minutes')`)
	if err != nil {
		return err
	}
	_, err = tx.Exec(ctx, `UPDATE fleet_firmware_download_ticket t SET revoked_at=now() FROM fleet_ota o WHERE o.id=t.ota_id AND o.status='timeout' AND t.revoked_at IS NULL`)
	if err != nil {
		return err
	}
	return tx.Commit(ctx)
}

// Manifest resolves the MQTT wakeup over authenticated HTTPS. The device must
// not install a URL or digest taken solely from the plaintext MQTT transport.
func (s *Service) Manifest(ctx context.Context, deviceID, taskID string) (map[string]any, error) {
	if _, err := uuid.Parse(taskID); err != nil {
		return nil, conflict("INVALID_OTA_TASK", "升级任务编号不正确")
	}
	tx, err := s.Pool.Begin(ctx)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback(ctx)
	var releaseID, model, version, digest string
	var size, code int64
	var created time.Time
	err = tx.QueryRow(ctx, `SELECT f.id::text,f.model,f.version,f.version_code,f.size,f.sha256,o.created_at
		FROM fleet_ota o JOIN fleet_firmware f ON f.id=o.release_id
		JOIN device d ON d.id=o.device_id AND d.tenant_id=f.tenant_id
		WHERE o.id=$1 AND o.device_id=$2 AND d.status='registered' AND d.activation_state NOT IN ('LOST','REVOKED')
		AND f.status='READY' AND f.model=d.model AND
		((o.status='pending' AND o.created_at>now()-interval '30 seconds') OR
		(o.status='accepted' AND o.created_at>now()-interval '20 minutes')) FOR UPDATE OF o`, taskID, deviceID).
		Scan(&releaseID, &model, &version, &code, &size, &digest, &created)
	if errors.Is(err, pgx.ErrNoRows) {
		return nil, conflict("OTA_TASK_UNAVAILABLE", "升级任务不可用或已过期")
	}
	if err != nil {
		return nil, err
	}
	_, hash, url, expires, err := newDownloadTicket(s.DownloadBaseURL)
	if err != nil {
		return nil, err
	}
	_, err = tx.Exec(ctx, `INSERT INTO fleet_firmware_download_ticket(token_hash,release_id,ota_id,expires_at) VALUES($1,$2,$3,$4)
		ON CONFLICT (ota_id) WHERE ota_id IS NOT NULL DO UPDATE SET token_hash=EXCLUDED.token_hash,expires_at=EXCLUDED.expires_at,revoked_at=NULL`, hash, releaseID, taskID, expires)
	if err != nil {
		return nil, err
	}
	if err = tx.Commit(ctx); err != nil {
		return nil, err
	}
	return map[string]any{"task_id": taskID, "model": model, "version": version, "version_code": code, "size": size, "sha256": digest, "download_url": url, "expires_at_ms": created.Add(30 * time.Second).UnixMilli()}, nil
}