using System.Text;
using Elsa.CustomServer.Nats;
using Elsa.Extensions;
using Elsa.Workflows;
using Elsa.Workflows.Attributes;
using Elsa.Workflows.Models;
using NATS.Client.Core;
using NATS.Client.JetStream;
using NATS.Client.JetStream.Models;

namespace Elsa.CustomServer.Activities;

[Activity("NATS", "NATS", "Publish a message to a NATS JetStream subject.", DisplayName = "NATS Publish", Kind = ActivityKind.Task)]
public class NatsPublish : CodeActivity
{
    [Input(
        DisplayName = "Connection string",
        Description =
            "NATS connection string. Format: <scheme>://[user:password@]host[:port]. " +
            "Schemes: nats, tls, ws, wss. " +
            "Examples: nats://demo.nats.io:4222 | nats://alice:s3cret@nats.example.com:4222 | " +
            "wss://alice:s3cret@nats.example.com/. " +
            "Special characters in the password must be URL-encoded (@ = %40, : = %3A, / = %2F, # = %23).",
        Category = "Connection")]
    public Input<string> Connection { get; set; } = default!;

    [Input(Description = "Subject to publish to (must be bound to a JetStream stream).")]
    public Input<string> Subject { get; set; } = default!;

    [Input(Description = "Message body (UTF-8 string).")]
    public Input<string> Body { get; set; } = default!;

    [Input(Description = "Optional Nats-Msg-Id used by JetStream for server-side deduplication within the stream's duplicate window.")]
    public Input<string?> MessageId { get; set; } = new(default(string?));

    [Input(Description = "Optional stream name for a fail-fast assertion that the subject lands in this stream.")]
    public Input<string?> ExpectedStream { get; set; } = new(default(string?));

    [Output(Description = "Stream sequence number the server assigned to the accepted message.")]
    public Output<long> Sequence { get; set; } = default!;

    [Output(Description = "True when JetStream deduplicated (Nats-Msg-Id already seen within the duplicate window).")]
    public Output<bool> Duplicate { get; set; } = default!;

    protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
    {
        var connection = context.Get(Connection);
        if (string.IsNullOrWhiteSpace(connection))
            throw new InvalidOperationException("NatsPublish: Connection string is required.");

        var subject = context.Get(Subject);
        if (string.IsNullOrWhiteSpace(subject))
            throw new InvalidOperationException("NatsPublish: Subject is required.");

        var body = context.Get(Body) ?? string.Empty;
        var messageId = context.Get(MessageId);
        var expectedStream = context.Get(ExpectedStream);

        var pool = context.GetRequiredService<NatsConnectionCache>();
        var nats = pool.Get(connection);
        var js = new NatsJSContext(nats);

        NatsHeaders? headers = null;
        if (!string.IsNullOrEmpty(messageId))
            headers = new NatsHeaders { ["Nats-Msg-Id"] = messageId };

        var opts = string.IsNullOrEmpty(expectedStream)
            ? null
            : new NatsJSPubOpts { ExpectedStream = expectedStream };

        var payload = Encoding.UTF8.GetBytes(body);
        var ack = await js.PublishAsync(subject, payload, headers: headers, opts: opts, cancellationToken: context.CancellationToken);
        ack.EnsureSuccess();

        context.Set(Sequence, (long)ack.Seq);
        context.Set(Duplicate, ack.Duplicate);
    }
}