65 lines
1.5 KiB
Go
65 lines
1.5 KiB
Go
package nats
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
)
|
|
|
|
type Client struct {
|
|
Conn *nats.Conn
|
|
JS jetstream.JetStream
|
|
}
|
|
|
|
// Connect initializes a NATS connection and JetStream context.
|
|
func Connect(url string) (*Client, error) {
|
|
nc, err := nats.Connect(url,
|
|
nats.Name("SimpleArmaAdmin"),
|
|
nats.MaxReconnects(-1), // Infinite retries
|
|
nats.ReconnectWait(2*time.Second),
|
|
nats.RetryOnFailedConnect(true), // Important for Docker startup
|
|
nats.Timeout(5*time.Second),
|
|
)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to connect to NATS: %w", err)
|
|
}
|
|
|
|
js, err := jetstream.New(nc)
|
|
if err != nil {
|
|
nc.Close()
|
|
return nil, fmt.Errorf("failed to create JetStream context: %w", err)
|
|
}
|
|
|
|
return &Client{
|
|
Conn: nc,
|
|
JS: js,
|
|
}, nil
|
|
}
|
|
|
|
// SetupStream ensures a JetStream stream exists for logs.
|
|
func (c *Client) SetupStream(ctx context.Context, streamName string, subjects []string) error {
|
|
_, err := c.JS.CreateOrUpdateStream(ctx, jetstream.StreamConfig{
|
|
Name: streamName,
|
|
Subjects: subjects,
|
|
Retention: jetstream.LimitsPolicy,
|
|
MaxAge: 30 * 24 * time.Hour, // 30 days retention as per blueprint
|
|
})
|
|
return err
|
|
}
|
|
|
|
// PublishLog sends an encrypted log blob to a specific subject.
|
|
func (c *Client) PublishLog(ctx context.Context, communityID string, logType string, data []byte) error {
|
|
subject := fmt.Sprintf("logs.%s.%s", communityID, logType)
|
|
_, err := c.JS.Publish(ctx, subject, data)
|
|
return err
|
|
}
|
|
|
|
func (c *Client) Close() {
|
|
if c.Conn != nil {
|
|
c.Conn.Close()
|
|
}
|
|
}
|