voyager-api/ScrapperAPI/Repositories/AgentRepository.cs

using Dapper;
using ScrapperAPI.Dtos;
using ScrapperAPI.Interfaces;

namespace ScrapperAPI.Repositories;

public sealed class AgentRepository : IAgentRepository
{
    private readonly IDbConnectionFactory _db;

    public AgentRepository(IDbConnectionFactory db) => _db = db;

    public async Task UpsertAsync(string agentId, string? displayName, string certThumbprint, CancellationToken ct)
    {
        const string sql = """
        insert into agent(id, display_name, cert_thumbprint, last_seen_at, is_enabled)
        values (@agentId, @displayName, @certThumbprint, now(), true)
        on conflict (id)
        do update set
            display_name = excluded.display_name,
            cert_thumbprint = excluded.cert_thumbprint,
            last_seen_at = now();
        """;

        using var conn = await _db.CreateOpenConnectionAsync(ct);
        await conn.ExecuteAsync(new CommandDefinition(sql, new { agentId, displayName, certThumbprint }, cancellationToken: ct));
    }

    public async Task<bool> IsEnabledAsync(string agentId, CancellationToken ct)
    {
        const string sql = """
        select is_enabled from agent where id = @agentId;
        """;
        using var conn = await _db.CreateOpenConnectionAsync(ct);
        return await conn.ExecuteScalarAsync<bool>(new CommandDefinition(sql, new { agentId }, cancellationToken: ct));
    }

    public async Task<string?> GetThumbprintAsync(string agentId, CancellationToken ct)
    {
        const string sql = """
        select cert_thumbprint from agent where id = @agentId;
        """;
        using var conn = await _db.CreateOpenConnectionAsync(ct);
        return await conn.ExecuteScalarAsync<string?>(new CommandDefinition(sql, new { agentId }, cancellationToken: ct));
    }

    public async Task TouchAsync(string agentId, CancellationToken ct)
    {
        const string sql = """
        update agent set last_seen_at = now() where id = @agentId;
        """;
        using var conn = await _db.CreateOpenConnectionAsync(ct);
        await conn.ExecuteAsync(new CommandDefinition(sql, new { agentId }, cancellationToken: ct));
    }

    public async Task<int> CountActiveAsync(TimeSpan seenWithin, CancellationToken ct)
    {
        const string sql = """
        select count(*)
        from agent
        where is_enabled = true
          and last_seen_at > now() - (@seenSeconds * interval '1 second');
        """;
        using var conn = await _db.CreateOpenConnectionAsync(ct);
        return await conn.ExecuteScalarAsync<int>(new CommandDefinition(sql, new { seenSeconds = (int)seenWithin.TotalSeconds }, cancellationToken: ct));
    }

    public async Task<AgentRow?> GetAsync(string agentId, CancellationToken ct)
    {
        const string sql = """
        select
            id as Id,
            display_name as DisplayName,
            cert_thumbprint as CertThumbprint,
            created_at as CreatedAt,
            last_seen_at as LastSeenAt,
            is_enabled as IsEnabled
        from agent
        where id = @agentId;
        """;
        using var conn = await _db.CreateOpenConnectionAsync(ct);
        return await conn.QuerySingleOrDefaultAsync<AgentRow>(new CommandDefinition(sql, new { agentId }, cancellationToken: ct));
    }
}