Codex Local
bf1233d5df
ci / verify (push) Canceled after 0s
Add customer and operator login, BLE provisioning, device fleet controls, call memory, memoir generation, and refreshed family and maintenance portals. Include API contracts, migrations, firmware sources, tests, assets, and deployment documentation.
324 lines
13 KiB
Go
324 lines
13 KiB
Go
// Package devicefleet implements the device MQTT heartbeat and explicit OTA workflow.
|
|
package devicefleet
|
|
|
|
import (
|
|
"context"
|
|
"crypto/hmac"
|
|
"crypto/sha256"
|
|
"encoding/base64"
|
|
"encoding/hex"
|
|
"errors"
|
|
"io"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgconn"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
type Error struct{ Code, Message string }
|
|
|
|
func (e *Error) Error() string { return e.Message }
|
|
func conflict(code, message string) error { return &Error{code, message} }
|
|
|
|
type Store interface {
|
|
FreezeFirmware(context.Context, string, string, string, int64) error
|
|
OpenDownload(context.Context, string) (io.ReadSeekCloser, error)
|
|
PresignUpload(context.Context, string, time.Duration) (string, error)
|
|
PresignMediaDownload(context.Context, string, time.Duration, string, string, bool) (string, error)
|
|
VerifyDigest(context.Context, string, string, int64) error
|
|
}
|
|
type Service struct {
|
|
Pool *pgxpool.Pool
|
|
Store Store
|
|
BrokerURL string
|
|
DownloadBaseURL string
|
|
Publish func(string, []byte) error
|
|
}
|
|
type Release struct {
|
|
ID string `json:"id"`
|
|
Model string `json:"model"`
|
|
Version string `json:"version"`
|
|
Code int64 `json:"version_code"`
|
|
Notes string `json:"release_notes"`
|
|
Size int64 `json:"size"`
|
|
SHA256 string `json:"sha256"`
|
|
Status string `json:"status"`
|
|
Created time.Time `json:"created_at"`
|
|
UploadURL string `json:"upload_url,omitempty"`
|
|
DownloadURL string `json:"download_url,omitempty"`
|
|
Object string `json:"-"`
|
|
}
|
|
type ReleaseInput struct {
|
|
Model string `json:"model"`
|
|
Version string `json:"version"`
|
|
Notes string `json:"release_notes"`
|
|
Size int64 `json:"size"`
|
|
SHA256 string `json:"sha256"`
|
|
}
|
|
|
|
const releaseFields = `id::text,model,version,version_code,release_notes,size,sha256,status,created_at,object_ref`
|
|
|
|
func scanRelease(row interface{ Scan(...any) error }) (r Release, err error) {
|
|
err = row.Scan(&r.ID, &r.Model, &r.Version, &r.Code, &r.Notes, &r.Size, &r.SHA256, &r.Status, &r.Created, &r.Object)
|
|
return
|
|
}
|
|
func (s *Service) Releases(ctx context.Context, tenant string) ([]Release, error) {
|
|
rows, err := s.Pool.Query(ctx, `SELECT `+releaseFields+` FROM fleet_firmware WHERE tenant_id=$1 ORDER BY version_code DESC LIMIT 200`, tenant)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
result := []Release{}
|
|
for rows.Next() {
|
|
r, e := scanRelease(rows)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
result = append(result, r)
|
|
}
|
|
return result, rows.Err()
|
|
}
|
|
func (s *Service) Models(ctx context.Context, tenant string) ([]string, error) {
|
|
rows, err := s.Pool.Query(ctx, `SELECT DISTINCT model FROM device WHERE tenant_id=$1 ORDER BY model`, tenant)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
result := []string{}
|
|
for rows.Next() {
|
|
var m string
|
|
if err = rows.Scan(&m); err != nil {
|
|
return nil, err
|
|
}
|
|
result = append(result, m)
|
|
}
|
|
return result, rows.Err()
|
|
}
|
|
func (s *Service) Prepare(ctx context.Context, tenant string, in ReleaseInput) (Release, error) {
|
|
in.Model = strings.TrimSpace(in.Model)
|
|
in.Version = strings.TrimSpace(in.Version)
|
|
in.SHA256 = strings.ToLower(in.SHA256)
|
|
hash, err := hex.DecodeString(in.SHA256)
|
|
if err != nil || len(hash) != 32 || in.Model == "" || len(in.Model) > 120 || in.Version == "" || len(in.Version) > 80 || len(in.Notes) > 8000 || in.Size <= 0 || in.Size > 32<<20 {
|
|
return Release{}, conflict("INVALID_FIRMWARE", "请选择型号，填写版本名称并上传不超过32MB的固件")
|
|
}
|
|
if s.Store == nil {
|
|
return Release{}, conflict("STORAGE_UNAVAILABLE", "固件存储未配置")
|
|
}
|
|
var exists bool
|
|
if err = s.Pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM device WHERE tenant_id=$1 AND model=$2)`, tenant, in.Model).Scan(&exists); err != nil {
|
|
return Release{}, err
|
|
}
|
|
if !exists {
|
|
return Release{}, conflict("MODEL_NOT_FOUND", "请先在设备库存登记该型号")
|
|
}
|
|
// Staging objects are never distributed. Finalize copies a verified immutable object.
|
|
object := "fleet/" + tenant + "/" + uuid.NewString() + ".bin"
|
|
url, err := s.Store.PresignUpload(ctx, object, 15*time.Minute)
|
|
if err != nil {
|
|
return Release{}, err
|
|
}
|
|
r, err := scanRelease(s.Pool.QueryRow(ctx, `INSERT INTO fleet_firmware(tenant_id,model,version,release_notes,size,sha256,object_ref) VALUES($1,$2,$3,$4,$5,$6,$7) ON CONFLICT(tenant_id,model,version) DO UPDATE SET release_notes=excluded.release_notes,size=excluded.size,sha256=excluded.sha256,object_ref=excluded.object_ref WHERE fleet_firmware.status='UPLOADING' RETURNING `+releaseFields, tenant, in.Model, in.Version, in.Notes, in.Size, in.SHA256, object))
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return Release{}, conflict("VERSION_EXISTS", "该型号已有同名已发布版本，请使用新的版本名称")
|
|
}
|
|
r.UploadURL = url
|
|
return r, err
|
|
}
|
|
func (s *Service) Finalize(ctx context.Context, tenant, id string) (Release, error) {
|
|
r, err := scanRelease(s.Pool.QueryRow(ctx, `SELECT `+releaseFields+` FROM fleet_firmware WHERE tenant_id=$1 AND id=$2`, tenant, id))
|
|
if err != nil {
|
|
return r, err
|
|
}
|
|
if r.Status == "READY" {
|
|
return r, nil
|
|
}
|
|
if s.Store == nil {
|
|
return r, conflict("STORAGE_UNAVAILABLE", "固件存储未配置")
|
|
}
|
|
// SHA-256 is included in every OTA command. Devices must verify it before booting.
|
|
finalObject := "fleet/ready/" + uuid.NewString() + ".bin"
|
|
if err = s.Store.FreezeFirmware(ctx, r.Object, finalObject, r.SHA256, r.Size); err != nil {
|
|
return r, conflict("FIRMWARE_INVALID", "文件大小或SHA256不匹配，请重新上传")
|
|
}
|
|
tag, err := s.Pool.Exec(ctx, `UPDATE fleet_firmware SET status='READY',object_ref=$3 WHERE id=$1 AND tenant_id=$2 AND status='UPLOADING' AND object_ref=$4`, id, tenant, finalObject, r.Object)
|
|
if err != nil {
|
|
return r, err
|
|
}
|
|
if tag.RowsAffected() != 1 {
|
|
return Release{}, conflict("UPLOAD_CHANGED", "上传状态已变化，请刷新后重试")
|
|
}
|
|
return scanRelease(s.Pool.QueryRow(ctx, `SELECT `+releaseFields+` FROM fleet_firmware WHERE id=$1 AND tenant_id=$2`, id, tenant))
|
|
}
|
|
func (s *Service) Download(ctx context.Context, tenant, id string) (Release, error) {
|
|
r, err := scanRelease(s.Pool.QueryRow(ctx, `SELECT `+releaseFields+` FROM fleet_firmware WHERE tenant_id=$1 AND id=$2 AND status='READY'`, tenant, id))
|
|
if err != nil {
|
|
return r, err
|
|
}
|
|
err = s.download(ctx, &r)
|
|
return r, err
|
|
}
|
|
func (s *Service) download(ctx context.Context, r *Release) error {
|
|
if s.Store == nil {
|
|
return conflict("STORAGE_UNAVAILABLE", "固件存储未配置")
|
|
}
|
|
_, tokenHash, address, expiresAt, err := newDownloadTicket(s.DownloadBaseURL)
|
|
if err != nil {
|
|
return conflict("DOWNLOAD_UNAVAILABLE", "固件短下载地址未正确配置")
|
|
}
|
|
if _, err = s.Pool.Exec(ctx, `INSERT INTO fleet_firmware_download_ticket(token_hash,release_id,expires_at) VALUES($1,$2,$3)`, tokenHash, r.ID, expiresAt); err != nil {
|
|
return err
|
|
}
|
|
r.DownloadURL = address
|
|
return nil
|
|
}
|
|
func Password(key []byte, factory string) string {
|
|
mac := hmac.New(sha256.New, key)
|
|
mac.Write([]byte("MQTT1\n" + factory))
|
|
return base64.RawURLEncoding.EncodeToString(mac.Sum(nil))
|
|
}
|
|
func StatusTopic(factory string) string { return "vl/devices/" + factory + "/status" }
|
|
func OTATopic(factory string) string { return "vl/devices/" + factory + "/ota" }
|
|
func (s *Service) Config(ctx context.Context, device, factory string, key []byte) (map[string]any, error) {
|
|
if s.BrokerURL == "" || s.Publish == nil {
|
|
return nil, conflict("MQTT_UNAVAILABLE", "MQTT服务未启用")
|
|
}
|
|
password := Password(key, factory)
|
|
hash := sha256.Sum256([]byte(password))
|
|
_, err := s.Pool.Exec(ctx, `INSERT INTO fleet_presence(device_id,password_hash) VALUES($1,$2) ON CONFLICT(device_id) DO UPDATE SET password_hash=excluded.password_hash`, device, hash[:])
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return map[string]any{"broker_url": s.BrokerURL, "client_id": factory, "username": factory, "password": password, "heartbeat_interval_seconds": 30, "keepalive_seconds": 60, "status_topic": StatusTopic(factory), "ota_topic": OTATopic(factory), "qos": 1}, nil
|
|
}
|
|
|
|
type Task struct {
|
|
ID string `json:"task_id"`
|
|
Status string `json:"status"`
|
|
Version string `json:"version"`
|
|
Code int64 `json:"version_code"`
|
|
Message string `json:"message"`
|
|
Created time.Time `json:"created_at"`
|
|
}
|
|
type State struct {
|
|
Online bool `json:"online"`
|
|
LastSeen *time.Time `json:"last_seen_at"`
|
|
CurrentVersion string `json:"current_version"`
|
|
CurrentCode int64 `json:"current_version_code"`
|
|
CurrentURL string `json:"current_download_url"`
|
|
OTASupported bool `json:"ota_supported"`
|
|
Latest *Release `json:"latest"`
|
|
UpdateAvailable bool `json:"update_available"`
|
|
Task *Task `json:"ota_task"`
|
|
}
|
|
|
|
func (s *Service) State(ctx context.Context, device string) (State, error) {
|
|
var out State
|
|
var tenant, model string
|
|
err := s.Pool.QueryRow(ctx, `SELECT d.tenant_id::text,d.model,COALESCE(p.connected AND p.last_seen_at>now()-interval '90 seconds',false),p.last_seen_at,COALESCE(NULLIF(p.firmware_version,''),dt.firmware_version,''),COALESCE(p.version_code,0),COALESCE(p.ota_supported,false) FROM device d LEFT JOIN fleet_presence p ON p.device_id=d.id LEFT JOIN device_twin dt ON dt.device_id=d.id WHERE d.id=$1`, device).Scan(&tenant, &model, &out.Online, &out.LastSeen, &out.CurrentVersion, &out.CurrentCode, &out.OTASupported)
|
|
if err != nil {
|
|
return out, err
|
|
}
|
|
latest, err := scanRelease(s.Pool.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 {
|
|
if err = s.download(ctx, &latest); err != nil {
|
|
return out, err
|
|
}
|
|
out.Latest = &latest
|
|
out.UpdateAvailable = latest.Code > out.CurrentCode
|
|
} else if !errors.Is(err, pgx.ErrNoRows) {
|
|
return out, err
|
|
}
|
|
current, err := scanRelease(s.Pool.QueryRow(ctx, `SELECT `+releaseFields+` FROM fleet_firmware WHERE tenant_id=$1 AND model=$2 AND version_code=$3 AND status='READY'`, tenant, model, out.CurrentCode))
|
|
if err == nil {
|
|
if err = s.download(ctx, ¤t); err != nil {
|
|
return out, err
|
|
}
|
|
out.CurrentURL = current.DownloadURL
|
|
} else if !errors.Is(err, pgx.ErrNoRows) {
|
|
return out, err
|
|
}
|
|
var t Task
|
|
err = s.Pool.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.device_id=$1 ORDER BY o.created_at DESC LIMIT 1`, device).Scan(&t.ID, &t.Status, &t.Version, &t.Code, &t.Message, &t.Created)
|
|
if err == nil {
|
|
out.Task = &t
|
|
} else if !errors.Is(err, pgx.ErrNoRows) {
|
|
return out, err
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
type Report struct {
|
|
TaskID string `json:"task_id"`
|
|
Status string `json:"status"`
|
|
FirmwareVersion string `json:"firmware_version"`
|
|
Code int64 `json:"version_code"`
|
|
Message string `json:"message,omitempty"`
|
|
}
|
|
|
|
func (s *Service) Report(ctx context.Context, device string, in Report) error {
|
|
if _, err := uuid.Parse(in.TaskID); err != nil {
|
|
return conflict("INVALID_REPORT", "升级任务ID不正确")
|
|
}
|
|
if in.Status != "accepted" && in.Status != "success" && in.Status != "failed" || len(in.Message) > 500 {
|
|
return conflict("INVALID_REPORT", "升级状态不正确")
|
|
}
|
|
tx, err := s.Pool.Begin(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback(ctx)
|
|
var status, version string
|
|
var code int64
|
|
var created time.Time
|
|
err = tx.QueryRow(ctx, `SELECT o.status,f.version,f.version_code,o.created_at FROM fleet_ota o JOIN fleet_firmware f ON f.id=o.release_id WHERE o.id=$1 AND o.device_id=$2 FOR UPDATE OF o`, in.TaskID, device).Scan(&status, &version, &code, &created)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return conflict("OTA_NOT_FOUND", "本设备没有该升级任务")
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if in.Status == "success" && (in.Code != code || in.FirmwareVersion != version) {
|
|
return conflict("VERSION_MISMATCH", "升级完成上报必须匹配目标版本和顺序号")
|
|
}
|
|
if status == in.Status {
|
|
return tx.Commit(ctx)
|
|
}
|
|
if in.Status == "accepted" && time.Since(created) > 30*time.Second {
|
|
return conflict("OTA_EXPIRED", "升级指令已过期，请刷新后重新发起")
|
|
}
|
|
if status == "success" || status == "failed" {
|
|
return conflict("OTA_FINISHED", "该升级任务已经结束")
|
|
}
|
|
if status == "timeout" && in.Status == "accepted" {
|
|
return conflict("OTA_EXPIRED", "升级指令已过期")
|
|
}
|
|
_, err = tx.Exec(ctx, `UPDATE fleet_ota SET status=$2,message=$3,updated_at=now() WHERE id=$1`, in.TaskID, in.Status, in.Message)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if in.Status == "success" || in.Status == "failed" {
|
|
if _, err = tx.Exec(ctx, `UPDATE fleet_firmware_download_ticket SET revoked_at=now() WHERE ota_id=$1 AND revoked_at IS NULL`, in.TaskID); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if in.Status == "success" {
|
|
var tag pgconn.CommandTag
|
|
tag, err = tx.Exec(ctx, `UPDATE fleet_presence SET firmware_version=$2,version_code=$3 WHERE device_id=$1 AND version_code<=$3`, device, version, code)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if tag.RowsAffected() == 0 {
|
|
return tx.Commit(ctx)
|
|
}
|
|
_, err = tx.Exec(ctx, `UPDATE device_twin SET firmware_version=$2,updated_at=now() WHERE device_id=$1`, device, version)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return tx.Commit(ctx)
|
|
}
|