76 lines
2.3 KiB
Go
76 lines
2.3 KiB
Go
package message
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
|
||
"git.asio.asia/nixevol/NixMsg/internal/app/port"
|
||
)
|
||
|
||
// OnHandshakeComplete 握手完成：写 online_since、清空不 keep 的 expire_at，并推送。
|
||
// 调用方须先把连接登记进 ConnRegistry（MemoryConns.Set）。
|
||
func (a *App) OnHandshakeComplete(ctx context.Context, endpointID string, conn LiveConn) error {
|
||
nowMs := a.now().UnixMilli()
|
||
err := a.db.Queue.Do(ctx, func(tx *sql.Tx) error {
|
||
if _, err := tx.Exec(`UPDATE endpoints SET online_since = ? WHERE id = ?`, nowMs, endpointID); err != nil {
|
||
return err
|
||
}
|
||
_, err := tx.Exec(`
|
||
UPDATE deliveries SET expire_at = NULL, updated_at = ?
|
||
WHERE endpoint_id = ? AND state = 'pending' AND keep = 0`, nowMs, endpointID)
|
||
return err
|
||
})
|
||
if err != nil {
|
||
return err
|
||
}
|
||
return a.PushPending(ctx, endpointID, conn.ConnID)
|
||
}
|
||
|
||
// OnDisconnect 连接断开：当前连接则延长宽限；按代号清 pushed_conn。
|
||
// 调用方负责从 ConnRegistry 移除连接。
|
||
func (a *App) OnDisconnect(ctx context.Context, endpointID string, connID port.ConnID, isCurrent bool) error {
|
||
nowMs := a.now().UnixMilli()
|
||
graceMs := a.lim.GraceSeconds * 1000
|
||
deadline := nowMs + graceMs
|
||
|
||
err := a.db.Queue.Do(ctx, func(tx *sql.Tx) error {
|
||
if isCurrent {
|
||
if _, err := tx.Exec(`UPDATE endpoints SET offline_since = ? WHERE id = ?`, nowMs, endpointID); err != nil {
|
||
return err
|
||
}
|
||
if _, err := tx.Exec(`
|
||
UPDATE deliveries SET
|
||
expire_at = CASE
|
||
WHEN expire_at IS NOT NULL AND expire_at > ? THEN expire_at
|
||
ELSE ?
|
||
END,
|
||
updated_at = ?
|
||
WHERE endpoint_id = ? AND state = 'pending' AND keep = 0`, deadline, deadline, nowMs, endpointID); err != nil {
|
||
return err
|
||
}
|
||
if _, err := tx.Exec(`
|
||
UPDATE deliveries SET
|
||
expire_at = CASE
|
||
WHEN expire_at IS NULL OR expire_at < ? THEN ?
|
||
ELSE expire_at
|
||
END,
|
||
updated_at = ?
|
||
WHERE endpoint_id = ? AND state = 'pending' AND keep = 1 AND pushed_conn = ?`,
|
||
deadline, deadline, nowMs, endpointID, string(connID)); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
_, err := tx.Exec(`
|
||
UPDATE deliveries SET pushed_conn = NULL, updated_at = ?
|
||
WHERE state = 'pending' AND pushed_conn = ?`, nowMs, string(connID))
|
||
return err
|
||
})
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if live, ok := a.lookupConn(endpointID); ok && live.ConnID != connID {
|
||
a.WakePush(endpointID)
|
||
}
|
||
return nil
|
||
}
|