justme
ad5efc62ff
Replace Probe with Agent across node runtime, Fleet contracts, backend state, API and storage. Report incoming sessions and outgoing link health through each Agent uplink, and retain directed link controls and retry. Update protocol fixtures and end-to-end coverage for the unified model; accept requested cancellation during the Fleet timeout test cleanup.
263 lines
15 KiB
C#
263 lines
15 KiB
C#
using Fleet.Core;
|
||
using Fleet.Server;
|
||
using Microsoft.Data.Sqlite;
|
||
using MoShiro.Backend.Application;
|
||
using MoShiro.Protocol;
|
||
|
||
namespace MoShiro.Backend.Storage.Sqlite;
|
||
|
||
/// <summary>
|
||
/// Agent 表同时是 Fleet 的节点表：实现 <see cref="IAgentStore"/>（MoShiro 的视图）和 <see cref="IFleetStore"/>（注册与认证）。
|
||
/// 加入令牌的校验也在这里，和 MoShiro 的其他数据放在同一个 moshiro.db 里，由同一个单写入者写入。
|
||
/// </summary>
|
||
public sealed class SqliteAgentStore : IAgentStore, IFleetStore
|
||
{
|
||
private const string AgentColumns =
|
||
"id, name, status, pubkey, tls_spki, public_host, tcp_port, quic_ports, labels, public_visible, kinds, agent_version, local_limits, " +
|
||
"task_version, applied_task_version, rejected, created_at, last_seen_at, row_version, ipv6_host, ipv6_supported, ipv6_diagnostics, public_host_manual, ipv6_host_manual, reported_addresses, tls_pubkey, added_by, listens";
|
||
|
||
private readonly SqliteDatabase _db;
|
||
|
||
public SqliteAgentStore(SqliteDatabase db)
|
||
{
|
||
_db = db;
|
||
}
|
||
|
||
// ───────────────────────── IAgentStore ─────────────────────────
|
||
|
||
public ValueTask<IReadOnlyList<AgentRecord>> ListAsync(CancellationToken ct) =>
|
||
_db.ReadAsync<IReadOnlyList<AgentRecord>>(c => c.Query($"SELECT {AgentColumns} FROM agents ORDER BY created_at, id", ReadAgent), ct);
|
||
|
||
public ValueTask<AgentRecord?> GetAsync(NodeId id, CancellationToken ct) =>
|
||
_db.ReadAsync(c => GetInternal(c, id), ct);
|
||
|
||
public ValueTask ApplyEnrollmentAsync(NodeId id, AgentEnrollment enrollment, CancellationToken ct) =>
|
||
_db.ExecuteAsync(c =>
|
||
{
|
||
c.Exec(
|
||
"UPDATE agents SET public_host = CASE WHEN public_host = '' THEN @host ELSE public_host END, tcp_port = @tcp, quic_ports = @quic, listens = @listens, " +
|
||
"kinds = CASE WHEN kinds = '[]' THEN @kinds ELSE kinds END, row_version = row_version + 1 WHERE id = @id",
|
||
("@id", id.Value), ("@host", enrollment.PublicHost), ("@tcp", enrollment.TcpPort), ("@listens", enrollment.TcpPort > 0 ? 1 : 0),
|
||
("@quic", StorageJson.ToJson(enrollment.QuicPorts)), ("@kinds", StorageJson.ToJson(enrollment.Kinds)));
|
||
}, ct);
|
||
|
||
public ValueTask<Result<AgentRecord>> PatchAsync(NodeId id, AgentPatch patch, long? expectedRowVersion, CancellationToken ct) =>
|
||
_db.WriteAsync(c =>
|
||
{
|
||
AgentRecord? current = GetInternal(c, id);
|
||
if (current is null)
|
||
{
|
||
return Result<AgentRecord>.Fail(MoShiroErrors.AgentNotFound, "agent not found");
|
||
}
|
||
|
||
if (expectedRowVersion is { } expected && expected != current.RowVersion)
|
||
{
|
||
return Result<AgentRecord>.Fail(MoShiroErrors.PreconditionFailed, "the agent was modified by someone else");
|
||
}
|
||
|
||
if (patch.Name is { } name && name != current.Name && current.Status != NodeStatus.Pending && NameTaken(c, name, id))
|
||
{
|
||
return Result<AgentRecord>.Fail(MoShiroErrors.NameConflict, "another agent already uses this name");
|
||
}
|
||
|
||
// 手填就标记为 manual（自动识别不再覆盖）；撤销手填回到自动，值留到下一次识别
|
||
bool hostAuto = patch.PublicHostAuto == true;
|
||
bool v6Auto = patch.Ipv6HostAuto == true;
|
||
c.Exec(
|
||
"UPDATE agents SET name = @name, labels = @labels, public_host = @host, ipv6_host = @v6, public_host_manual = @hostManual, ipv6_host_manual = @v6Manual, " +
|
||
"public_visible = @visible, row_version = row_version + 1 WHERE id = @id",
|
||
("@id", id.Value), ("@name", patch.Name ?? current.Name),
|
||
("@labels", StorageJson.ToJson(patch.Labels ?? current.Labels)),
|
||
("@host", hostAuto ? current.PublicHost : patch.PublicHost ?? current.PublicHost), ("@v6", v6Auto ? current.Ipv6Host : patch.Ipv6Host ?? current.Ipv6Host),
|
||
("@hostManual", hostAuto ? 0 : patch.PublicHost is not null || current.PublicHostManual ? 1 : 0),
|
||
("@v6Manual", v6Auto ? 0 : patch.Ipv6Host is not null || current.Ipv6HostManual ? 1 : 0),
|
||
("@visible", (patch.PublicVisible ?? current.PublicVisible) ? 1 : 0));
|
||
return Result<AgentRecord>.Ok(GetInternal(c, id)!);
|
||
}, ct);
|
||
|
||
public ValueTask<Result<AgentRecord>> SetStatusAsync(NodeId id, NodeStatus status, CancellationToken ct) =>
|
||
_db.WriteAsync(c =>
|
||
{
|
||
AgentRecord? current = GetInternal(c, id);
|
||
if (current is null)
|
||
{
|
||
return Result<AgentRecord>.Fail(MoShiroErrors.AgentNotFound, "agent not found");
|
||
}
|
||
|
||
if (status != NodeStatus.Pending && NameTaken(c, current.Name, id))
|
||
{
|
||
return Result<AgentRecord>.Fail(MoShiroErrors.NameConflict, "another agent already uses this name; rename it or take over the existing node");
|
||
}
|
||
|
||
c.Exec("UPDATE agents SET status = @status, row_version = row_version + 1 WHERE id = @id", ("@id", id.Value), ("@status", StatusText(status)));
|
||
return Result<AgentRecord>.Ok(GetInternal(c, id)!);
|
||
}, ct);
|
||
|
||
public ValueTask<Result<AgentRecord>> TakeoverAsync(NodeId pendingId, CancellationToken ct) =>
|
||
_db.WriteAsync(c =>
|
||
{
|
||
AgentRecord? pending = GetInternal(c, pendingId);
|
||
if (pending is null || pending.Status != NodeStatus.Pending)
|
||
{
|
||
return Result<AgentRecord>.Fail(MoShiroErrors.AgentNotFound, "no pending agent with this id");
|
||
}
|
||
|
||
string? existingId = c.Scalar<string>("SELECT id FROM agents WHERE name = @name AND status <> 'pending' AND id <> @id", ("@name", pending.Name), ("@id", pendingId.Value));
|
||
if (existingId is null)
|
||
{
|
||
return Result<AgentRecord>.Fail(MoShiroErrors.AgentNotFound, "there is no established agent with this name to take over");
|
||
}
|
||
|
||
// 旧记录换成新公钥，历史数据保留；待审批记录删除
|
||
c.Exec("DELETE FROM agents WHERE id = @id", ("@id", pendingId.Value));
|
||
c.Exec(
|
||
"UPDATE agents SET pubkey = @pubkey, tls_spki = @spki, tls_pubkey = @peer, status = 'active', task_hash = NULL, row_version = row_version + 1 WHERE id = @id",
|
||
("@id", existingId), ("@pubkey", pending.PublicKey), ("@spki", pending.TlsSpki), ("@peer", pending.TlsPublicKey));
|
||
return Result<AgentRecord>.Ok(GetInternal(c, new NodeId(existingId))!);
|
||
}, ct);
|
||
|
||
public ValueTask DeleteAsync(NodeId id, CancellationToken ct) =>
|
||
_db.ExecuteAsync(c => c.Exec("DELETE FROM agents WHERE id = @id", ("@id", id.Value)), ct);
|
||
|
||
public ValueTask ApplyHelloAsync(NodeId id, AgentHello hello, CancellationToken ct) =>
|
||
_db.ExecuteAsync(c =>
|
||
{
|
||
ListenInfo? listen = hello.Listen;
|
||
c.Exec(
|
||
"UPDATE agents SET kinds = @kinds, agent_version = @version, local_limits = COALESCE(@limits, local_limits), listens = @listens, " +
|
||
"tcp_port = COALESCE(@tcp, tcp_port), quic_ports = COALESCE(@quic, quic_ports), ipv6_supported = @v6, ipv6_diagnostics = @diag, " +
|
||
"reported_addresses = COALESCE(@addr, reported_addresses), row_version = row_version + 1 WHERE id = @id",
|
||
("@id", id.Value), ("@kinds", StorageJson.ToJson(hello.Kinds)), ("@version", hello.AgentVersion), ("@limits", hello.LocalLimitsJson),
|
||
("@listens", listen is null ? 0 : 1), ("@tcp", listen?.TcpPort), ("@quic", listen is null ? null : StorageJson.ToJson(listen.QuicPorts)),
|
||
("@v6", listen?.Ipv6Supported == true ? 1 : 0), ("@diag", hello.Ipv6DiagnosticsJson),
|
||
("@addr", listen?.Addresses is null ? null : StorageJson.ToJson(listen.Addresses)));
|
||
}, ct);
|
||
|
||
public ValueTask ApplyAutoAddressesAsync(NodeId id, string? publicHost, string? ipv6Host, CancellationToken ct) =>
|
||
_db.ExecuteAsync(c =>
|
||
{
|
||
c.Exec(
|
||
"UPDATE agents SET public_host = CASE WHEN public_host_manual = 0 AND @host IS NOT NULL THEN @host ELSE public_host END, " +
|
||
"ipv6_host = CASE WHEN ipv6_host_manual = 0 AND @v6 IS NOT NULL THEN @v6 ELSE ipv6_host END, row_version = row_version + 1 " +
|
||
"WHERE id = @id AND ((public_host_manual = 0 AND @host IS NOT NULL AND public_host <> @host) OR (ipv6_host_manual = 0 AND @v6 IS NOT NULL AND ipv6_host <> @v6))",
|
||
("@id", id.Value), ("@host", publicHost), ("@v6", ipv6Host));
|
||
}, ct);
|
||
|
||
public ValueTask SetIpv6DiagnosticsAsync(NodeId id, string diagnosticsJson, CancellationToken ct) =>
|
||
_db.ExecuteAsync(c => c.Exec("UPDATE agents SET ipv6_diagnostics = @diag WHERE id = @id", ("@diag", diagnosticsJson), ("@id", id.Value)), ct);
|
||
|
||
public ValueTask SetTaskVersionAsync(NodeId id, long version, string taskHash, CancellationToken ct) =>
|
||
_db.ExecuteAsync(c => c.Exec("UPDATE agents SET task_version = @v, task_hash = @h WHERE id = @id", ("@id", id.Value), ("@v", version), ("@h", taskHash)), ct);
|
||
|
||
public ValueTask<string?> GetTaskHashAsync(NodeId id, CancellationToken ct) =>
|
||
_db.ReadAsync(c => c.Scalar<string>("SELECT task_hash FROM agents WHERE id = @id", ("@id", id.Value)), ct);
|
||
|
||
public ValueTask SetAppliedTaskStateAsync(NodeId id, long appliedVersion, string? rejectedJson, CancellationToken ct) =>
|
||
_db.ExecuteAsync(c => c.Exec("UPDATE agents SET applied_task_version = @v, rejected = @r WHERE id = @id", ("@id", id.Value), ("@v", appliedVersion), ("@r", rejectedJson)), ct);
|
||
|
||
// ───────────────────────── IFleetStore ─────────────────────────
|
||
|
||
public ValueTask<JoinTokenRecord?> FindJoinTokenAsync(ReadOnlyMemory<byte> tokenSha256, CancellationToken ct)
|
||
{
|
||
byte[] hash = tokenSha256.ToArray();
|
||
return _db.ReadAsync(
|
||
c => c.QueryOne(
|
||
"SELECT id, token_sha256, note, labels, auto_approve, max_uses, uses, expires_at, revoked_at, created_at, kind, created_by FROM join_tokens WHERE token_sha256 = @h",
|
||
r => new JoinTokenRecord(r.GetString(0), r.Bytes(1), r.NullableString(2), StorageJson.Strings(r.GetString(3)), r.GetInt64(4) != 0, r.GetInt32(5), r.GetInt32(6), r.GetInt64(7), r.NullableLong(8), r.GetInt64(9), r.GetString(10), r.NullableString(11)),
|
||
("@h", hash)),
|
||
ct);
|
||
}
|
||
|
||
public ValueTask<bool> TryConsumeJoinTokenAsync(string tokenId, long nowMs, CancellationToken ct) =>
|
||
_db.WriteAsync(
|
||
c => c.Exec(
|
||
"UPDATE join_tokens SET uses = uses + 1 WHERE id = @id AND revoked_at IS NULL AND expires_at > @now AND uses < max_uses",
|
||
("@id", tokenId), ("@now", nowMs)) == 1,
|
||
ct);
|
||
|
||
public ValueTask<Result<NodeRecord>> CreateNodeAsync(NewNode node, CancellationToken ct) =>
|
||
_db.WriteAsync(c =>
|
||
{
|
||
if (node.Status != NodeStatus.Pending && NameTaken(c, node.Name, node.Id))
|
||
{
|
||
return Result<NodeRecord>.Fail(FleetErrors.NameConflict, "a node with this name already exists");
|
||
}
|
||
|
||
c.Exec(
|
||
"INSERT INTO agents (id, name, status, pubkey, tls_spki, tls_pubkey, labels, meta, agent_version, created_at, added_by) VALUES (@id, @name, @status, @pubkey, @spki, @peer, @labels, @meta, @version, @now, @by)",
|
||
("@id", node.Id.Value), ("@name", node.Name), ("@status", StatusText(node.Status)), ("@pubkey", node.PublicKey), ("@spki", node.TlsSpki), ("@peer", node.TlsPublicKey),
|
||
("@labels", StorageJson.ToJson(node.Labels)), ("@meta", node.MetaJson), ("@version", node.NodeVersion), ("@now", node.NowMs), ("@by", node.AddedBy));
|
||
return Result<NodeRecord>.Ok(ToNode(GetInternal(c, node.Id)!, node.MetaJson));
|
||
}, ct);
|
||
|
||
ValueTask<NodeRecord?> IFleetStore.GetNodeAsync(NodeId id, CancellationToken ct) =>
|
||
_db.ReadAsync(c => GetInternal(c, id) is { } agent ? ToNode(agent, null) : null, ct);
|
||
|
||
public ValueTask<NodeRecord?> FindEstablishedNodeByNameAsync(string name, CancellationToken ct) =>
|
||
_db.ReadAsync(
|
||
c => c.QueryOne($"SELECT {AgentColumns} FROM agents WHERE name = @name AND status <> 'pending'", ReadAgent, ("@name", name)) is { } agent ? ToNode(agent, null) : null,
|
||
ct);
|
||
|
||
public ValueTask UpdatePresenceAsync(NodeId id, Presence presence, CancellationToken ct) =>
|
||
_db.ExecuteAsync(
|
||
c => c.Exec(
|
||
"UPDATE agents SET last_seen_at = @at, agent_version = COALESCE(@version, agent_version) WHERE id = @id",
|
||
("@id", id.Value), ("@at", presence.AtMs), ("@version", presence.NodeVersion)),
|
||
ct);
|
||
|
||
// ───────────────────────── 内部 ─────────────────────────
|
||
|
||
private static AgentRecord? GetInternal(SqliteConnection c, NodeId id) =>
|
||
c.QueryOne($"SELECT {AgentColumns} FROM agents WHERE id = @id", ReadAgent, ("@id", id.Value));
|
||
|
||
private static bool NameTaken(SqliteConnection c, string name, NodeId except) =>
|
||
c.Scalar<long>("SELECT COUNT(*) FROM agents WHERE name = @name AND status <> 'pending' AND id <> @id", ("@name", name), ("@id", except.Value)) > 0;
|
||
|
||
private static AgentRecord ReadAgent(SqliteDataReader r) => new(
|
||
new NodeId(r.GetString(0)),
|
||
r.GetString(1),
|
||
ParseStatus(r.GetString(2)),
|
||
r.Bytes(3),
|
||
r.Bytes(4),
|
||
r.GetString(5),
|
||
r.GetInt32(6),
|
||
StorageJson.Ints(r.GetString(7)),
|
||
StorageJson.Strings(r.GetString(8)),
|
||
r.GetInt64(9) != 0,
|
||
StorageJson.Strings(r.GetString(10)),
|
||
r.NullableString(11),
|
||
r.NullableString(12),
|
||
r.GetInt64(13),
|
||
r.GetInt64(14),
|
||
r.NullableString(15),
|
||
r.GetInt64(16),
|
||
r.NullableLong(17),
|
||
r.GetInt64(18),
|
||
r.GetString(19),
|
||
r.GetInt64(20) != 0,
|
||
r.NullableString(21),
|
||
r.GetInt64(22) != 0,
|
||
r.GetInt64(23) != 0,
|
||
StorageJson.Strings(r.GetString(24)),
|
||
r.NullableBytes(25),
|
||
r.NullableString(26),
|
||
r.GetInt64(27) != 0);
|
||
|
||
private static NodeRecord ToNode(AgentRecord a, string? metaJson) =>
|
||
new(a.Id, a.Name, a.Status, a.PublicKey, a.TlsSpki, a.Labels, metaJson, a.AgentVersion, a.CreatedAtMs, a.LastSeenAtMs);
|
||
|
||
private static string StatusText(NodeStatus status) => status switch
|
||
{
|
||
NodeStatus.Active => "active",
|
||
NodeStatus.Disabled => "disabled",
|
||
_ => "pending",
|
||
};
|
||
|
||
private static NodeStatus ParseStatus(string text) => text switch
|
||
{
|
||
"active" => NodeStatus.Active,
|
||
"disabled" => NodeStatus.Disabled,
|
||
_ => NodeStatus.Pending,
|
||
};
|
||
}
|