V2 upgrade

NostrStreamer/ApiModel/Account.cs

@@ -5,20 +5,44 @@ namespace NostrStreamer.ApiModel;
 
 public class Account
 {
+    [JsonProperty("event")]
+    public NostrEvent? Event { get; init; }
+
+    [JsonProperty("endpoints")]
+    public List<AccountEndpoint> Endpoints { get; init; } = new();
+    
+    [JsonProperty("balance")]
+    public long Balance { get; init; }
+}
+
+public class AccountEndpoint
+{
+    [JsonProperty("name")]
+    public string Name { get; init; } = null!;
+    
     [JsonProperty("url")]
     public string Url { get; init; } = null!;
     
     [JsonProperty("key")]
     public string Key { get; init; } = null!;
 
-    [JsonProperty("event")]
+    [JsonProperty("cost")]
-    public NostrEvent? Event { get; init; }
+    public EndpointCost Cost { get; init; } = null!;
 
-    [JsonProperty("quota")]
+    [JsonProperty("capabilities")]
-    public AccountQuota Quota { get; init; } = null!;
+    public List<string> Capabilities { get; init; } = new();
 }
 
+public class EndpointCost
+{
+    [JsonProperty("rate")]
+    public double Rate { get; init; }
 
+    [JsonProperty("unit")]
+    public string Unit { get; init; } = null!;
+}
+
+[Obsolete("Use EndpointCost")]
 public class AccountQuota
 {
     [JsonProperty("rate")]

NostrStreamer/Config.cs

@@ -12,11 +12,6 @@ public class Config
     /// </summary>
     public Uri RtmpHost { get; init; } = null!;
 
-    /// <summary>
-    /// SRS app name
-    /// </summary>
-    public string App { get; init; } = "live";
-
     /// <summary>
     /// SRS api server host
     /// </summary>
@@ -32,20 +27,15 @@ public class Config
     /// </summary>
     public Uri DataHost { get; init; } = null!;
 
+    /// <summary>
+    /// Public URL for the api
+    /// </summary>
+    public Uri ApiHost { get; init; } = null!;
+
     public string PrivateKey { get; init; } = null!;
     public string[] Relays { get; init; } = Array.Empty<string>();
 
     public LndConfig Lnd { get; init; } = null!;
-
-    /// <summary>
-    /// Cost/min (milli-sats)
-    /// </summary>
-    public int Cost { get; init; } = 10_000;
-
-    /// <summary>
-    /// List of video variants
-    /// </summary>
-    public List<Variant> Variants { get; init; } = null!;
 }
 
 public class LndConfig
@@ -56,11 +46,3 @@ public class LndConfig
 
     public string MacaroonPath { get; init; } = null!;
 }
-
-public class Variant
-{
-    public string Name { get; init; } = null!;
-    public int Width { get; init; }
-    public int Height { get; init; }
-    public int Bandwidth { get; init; }
-}

NostrStreamer/Controllers/LnurlController.cs

@@ -0,0 +1,86 @@
+using BTCPayServer.Lightning;
+using LNURL;
+using Microsoft.AspNetCore.Mvc;
+using Newtonsoft.Json;
+using Nostr.Client.Json;
+using Nostr.Client.Messages;
+using NostrStreamer.Database;
+using NostrStreamer.Services;
+
+namespace NostrStreamer.Controllers;
+
+[Route("/api/pay")]
+public class LnurlController : Controller
+{
+    private readonly Config _config;
+    private readonly UserService _userService;
+
+    public LnurlController(Config config, UserService userService)
+    {
+        _config = config;
+        _userService = userService;
+    }
+
+    [HttpGet("/.well-known/lnurlp/{key}")]
+    public async Task<IActionResult> GetPayService([FromRoute] string key)
+    {
+        var user = await _userService.GetUser(key);
+        if (user == default) return LnurlError("User not found");
+
+        var metadata = GetMetadata(user);
+        var pubKey = _config.GetPubKey();
+        return Json(new LNURLPayRequest
+        {
+            Callback = new Uri(_config.ApiHost, $"/api/pay/{key}"),
+            Metadata = JsonConvert.SerializeObject(metadata),
+            MinSendable = LightMoney.Satoshis(1),
+            MaxSendable = LightMoney.Coins(1),
+            Tag = "payRequest",
+            NostrPubkey = pubKey
+        });
+    }
+
+    [HttpGet("{key}")]
+    public async Task<IActionResult> PayUserBalance([FromRoute] string key, [FromQuery] long amount, [FromQuery] string? nostr)
+    {
+        try
+        {
+            if (!string.IsNullOrEmpty(nostr))
+            {
+                var ev = JsonConvert.DeserializeObject<NostrEvent>(nostr, NostrSerializer.Settings);
+                if (ev?.Kind != NostrKind.ZapRequest || ev.Tags?.FindFirstTagValue("amount") != amount.ToString() ||
+                    !ev.IsSignatureValid())
+                {
+                    throw new Exception("Invalid nostr event");
+                }
+            }
+
+            var invoice = await _userService.CreateTopup(key, (ulong)(amount / 1000), nostr);
+            return Json(new LNURLPayRequest.LNURLPayRequestCallbackResponse
+            {
+                Pr = invoice
+            });
+        }
+        catch (Exception ex)
+        {
+            return LnurlError($"Failed to create invoice (${ex.Message})");
+        }
+    }
+
+    private List<KeyValuePair<string, string>> GetMetadata(User u)
+    {
+        return new List<KeyValuePair<string, string>>()
+        {
+            new("text/plain", $"Topup for {u.PubKey}")
+        };
+    }
+
+    private IActionResult LnurlError(string reason)
+    {
+        return Json(new LNUrlStatusResponse()
+        {
+            Reason = reason,
+            Status = "ERROR"
+        });
+    }
+}

NostrStreamer/Controllers/NostrController.cs

@@ -4,11 +4,11 @@ using Microsoft.AspNetCore.Mvc;
 using Microsoft.EntityFrameworkCore;
 using Newtonsoft.Json;
 using Nostr.Client.Json;
-using Nostr.Client.Messages;
 using Nostr.Client.Utils;
 using NostrStreamer.ApiModel;
 using NostrStreamer.Database;
 using NostrStreamer.Services;
+using NostrStreamer.Services.StreamManager;
 
 namespace NostrStreamer.Controllers;
 
@@ -18,14 +18,14 @@ public class NostrController : Controller
 {
     private readonly StreamerContext _db;
     private readonly Config _config;
-    private readonly StreamManager _streamManager;
+    private readonly StreamManagerFactory _streamManagerFactory;
     private readonly LndNode _lnd;
 
-    public NostrController(StreamerContext db, Config config, StreamManager streamManager, LndNode lnd)
+    public NostrController(StreamerContext db, Config config, StreamManagerFactory streamManager, LndNode lnd)
     {
         _db = db;
         _config = config;
-        _streamManager = streamManager;
+        _streamManagerFactory = streamManager;
         _lnd = lnd;
     }
 
@@ -48,18 +48,23 @@ public class NostrController : Controller
             await _db.SaveChangesAsync();
         }
 
+        var endpoints = await _db.Endpoints.ToListAsync();
         var account = new Account
         {
-            Url = new Uri(_config.RtmpHost, _config.App).ToString(),
+            Event = null,
+            Endpoints = endpoints.Select(a => new AccountEndpoint()
+            {
+                Name = a.Name,
+                Url = new Uri(_config.RtmpHost, a.App).ToString(),
                 Key = user.StreamKey,
-            Event = !string.IsNullOrEmpty(user.Event) ? JsonConvert.DeserializeObject<NostrEvent>(user.Event, NostrSerializer.Settings) :
+                Capabilities = a.Capabilities,
-                null,
+                Cost = new()
-            Quota = new()
                 {
                     Unit = "min",
-                Rate = (int)Math.Ceiling(_config.Cost / 1000m),
+                    Rate = a.Cost / 1000d
-                Remaining = (long)Math.Floor(user.Balance / 1000m)
                 }
+            }).ToList(),
+            Balance = (long)Math.Floor(user.Balance / 1000m)
         };
 
         return Content(JsonConvert.SerializeObject(account, NostrSerializer.Settings), "application/json");
@@ -71,7 +76,8 @@ public class NostrController : Controller
         var pubkey = GetPubKey();
         if (string.IsNullOrEmpty(pubkey)) return Unauthorized();
 
-        await _streamManager.PatchEvent(pubkey, req.Title, req.Summary, req.Image, req.Tags, req.ContentWarning);
+        var streamManager = await _streamManagerFactory.ForCurrentStream(pubkey);
+        await streamManager.PatchEvent(req.Title, req.Summary, req.Image, req.Tags, req.ContentWarning);
         return Accepted();
     }
 

NostrStreamer/Controllers/PlaylistController.cs

@@ -1,9 +1,8 @@
 using System.Text.RegularExpressions;
 using Microsoft.AspNetCore.Mvc;
-using Microsoft.EntityFrameworkCore;
-using Microsoft.Extensions.Caching.Memory;
 using NostrStreamer.Database;
 using NostrStreamer.Services;
+using NostrStreamer.Services.StreamManager;
 
 namespace NostrStreamer.Controllers;
 
@@ -11,36 +10,30 @@ namespace NostrStreamer.Controllers;
 public class PlaylistController : Controller
 {
     private readonly ILogger<PlaylistController> _logger;
-    private readonly IMemoryCache _cache;
     private readonly Config _config;
-    private readonly IServiceScopeFactory _scopeFactory;
     private readonly HttpClient _client;
     private readonly SrsApi _srsApi;
     private readonly ViewCounter _viewCounter;
+    private readonly StreamManagerFactory _streamManagerFactory;
 
-    public PlaylistController(Config config, IMemoryCache cache, ILogger<PlaylistController> logger, IServiceScopeFactory scopeFactory,
+    public PlaylistController(Config config, ILogger<PlaylistController> logger,
-        HttpClient client, SrsApi srsApi, ViewCounter viewCounter)
+        HttpClient client, SrsApi srsApi, ViewCounter viewCounter, StreamManagerFactory streamManagerFactory)
     {
         _config = config;
-        _cache = cache;
         _logger = logger;
-        _scopeFactory = scopeFactory;
         _client = client;
         _srsApi = srsApi;
         _viewCounter = viewCounter;
+        _streamManagerFactory = streamManagerFactory;
     }
 
-    [HttpGet("{variant}/{pubkey}.m3u8")]
+    [HttpGet("{variant}/{id}.m3u8")]
-    public async Task RewritePlaylist([FromRoute] string pubkey, [FromRoute] string variant, [FromQuery(Name = "hls_ctx")] string hlsCtx)
+    public async Task RewritePlaylist([FromRoute] Guid id, [FromRoute] string variant, [FromQuery(Name = "hls_ctx")] string hlsCtx)
     {
-        var key = await GetStreamKey(pubkey);
+        var streamManager = await _streamManagerFactory.ForStream(id);
-        if (string.IsNullOrEmpty(key))
+        var userStream = streamManager.GetStream();
-        {
-            Response.StatusCode = 404;
-            return;
-        }
 
-        var path = $"/{_config.App}/{variant}/{key}.m3u8";
+        var path = $"/{userStream.Endpoint.App}/{variant}/{userStream.User.StreamKey}.m3u8";
         var ub = new UriBuilder(_config.SrsHttpHost)
         {
             Path = path,
@@ -66,8 +59,8 @@ public class PlaylistController : Controller
             {
                 await sw.WriteLineAsync(line);
                 var trackPath = await sr.ReadLineAsync();
-                var seg = Regex.Match(trackPath!, @"-(\d+)\.ts$");
+                var seg = Regex.Match(trackPath!, @"-(\d+)\.ts");
-                await sw.WriteLineAsync($"{pubkey}/{seg.Groups[1].Value}.ts");
+                await sw.WriteLineAsync($"{id}/{seg.Groups[1].Value}.ts");
             }
             else
             {
@@ -76,20 +69,18 @@ public class PlaylistController : Controller
         }
 
         Response.Body.Close();
-        _viewCounter.Activity(key, hlsCtx);
+        _viewCounter.Activity(userStream.Id, hlsCtx);
     }
 
     [HttpGet("{pubkey}.m3u8")]
     public async Task CreateMultiBitrate([FromRoute] string pubkey)
     {
-        var key = await GetStreamKey(pubkey);
+        try
-        if (string.IsNullOrEmpty(key))
         {
-            Response.StatusCode = 404;
+            var streamManager = await _streamManagerFactory.ForCurrentStream(pubkey);
-            return;
-        }
 
-        var hlsCtx = await GetHlsCtx(key);
+            var userStream = streamManager.GetStream();
+            var hlsCtx = await GetHlsCtx(userStream);
             if (string.IsNullOrEmpty(hlsCtx))
             {
                 Response.StatusCode = 404;
@@ -102,45 +93,48 @@ public class PlaylistController : Controller
             var streams = await _srsApi.ListStreams();
             await sw.WriteLineAsync("#EXTM3U");
 
-        foreach (var variant in _config.Variants.OrderBy(a => a.Bandwidth))
+            foreach (var variant in userStream.Endpoint.GetVariants().OrderBy(a => a.Bandwidth))
             {
                 var stream = streams.FirstOrDefault(a =>
-                a.Name == key && a.App == $"{_config.App}/{variant.Name}");
+                    a.Name == userStream.User.StreamKey && a.App == $"{userStream.Endpoint.App}/{variant.SourceName}");
 
                 var resArg = stream?.Video != default ? $"RESOLUTION={stream.Video?.Width}x{stream.Video?.Height}" :
-                $"RESOLUTION={variant.Width}x{variant.Height}";
+                    variant.ToResolutionArg();
 
-            var bandwidthArg = $"BANDWIDTH={variant.Bandwidth * 1000}";
+                var bandwidthArg = variant.ToBandwidthArg();
 
                 var averageBandwidthArg = stream?.Kbps?.Recv30s.HasValue ?? false ? $"AVERAGE-BANDWIDTH={stream.Kbps.Recv30s * 1000}" : "";
-            var allArgs = new[] {bandwidthArg, averageBandwidthArg, resArg}.Where(a => !string.IsNullOrEmpty(a));
+                var codecArg = "CODECS=\"avc1.640028,mp4a.40.2\"";
+                var allArgs = new[] {bandwidthArg, averageBandwidthArg, resArg, codecArg}.Where(a => !string.IsNullOrEmpty(a));
                 await sw.WriteLineAsync(
-                $"#EXT-X-STREAM-INF:{string.Join(",", allArgs)},CODECS=\"avc1.640028,mp4a.40.2\"");
+                    $"#EXT-X-STREAM-INF:{string.Join(",", allArgs)}");
 
                 var u = new Uri(_config.DataHost,
-                $"{variant.Name}/{pubkey}.m3u8{(!string.IsNullOrEmpty(hlsCtx) ? $"?hls_ctx={hlsCtx}" : "")}");
+                    $"{variant.SourceName}/{userStream.Id}.m3u8{(!string.IsNullOrEmpty(hlsCtx) ? $"?hls_ctx={hlsCtx}" : "")}");
 
                 await sw.WriteLineAsync(u.ToString());
             }
         }
-
+        catch (Exception ex)
-    [HttpGet("{variant}/{pubkey}/{segment}")]
-    public async Task GetSegment([FromRoute] string pubkey, [FromRoute] string segment, [FromRoute] string variant)
-    {
-        var key = await GetStreamKey(pubkey);
-        if (string.IsNullOrEmpty(key))
         {
+            _logger.LogWarning(ex, "Failed to get stream for {pubkey} {message}", pubkey, ex.Message);
             Response.StatusCode = 404;
-            return;
+        }
     }
 
-        var path = $"/{_config.App}/{variant}/{key}-{segment}";
+    [HttpGet("{variant}/{id}/{segment}")]
+    public async Task GetSegment([FromRoute] Guid id, [FromRoute] string segment, [FromRoute] string variant)
+    {
+        var streamManager = await _streamManagerFactory.ForStream(id);
+        var userStream = streamManager.GetStream();
+
+        var path = $"/{userStream.Endpoint.App}/{variant}/{userStream.User.StreamKey}-{segment}";
         await ProxyRequest(path);
     }
 
-    private async Task<string?> GetHlsCtx(string key)
+    private async Task<string?> GetHlsCtx(UserStream stream)
     {
-        var path = $"/{_config.App}/source/{key}.m3u8";
+        var path = $"/{stream.Endpoint.App}/source/{stream.User.StreamKey}.m3u8";
         var ub = new Uri(_config.SrsHttpHost, path);
         var req = CreateProxyRequest(ub);
         using var rsp = await _client.SendAsync(req);
@@ -186,21 +180,4 @@ public class PlaylistController : Controller
 
         return req;
     }
-
-    private async Task<string?> GetStreamKey(string pubkey)
-    {
-        var cacheKey = $"stream-key:{pubkey}";
-        var cached = _cache.Get<string>(cacheKey);
-        if (cached != default)
-        {
-            return cached;
-        }
-
-        using var scope = _scopeFactory.CreateScope();
-        await using var db = scope.ServiceProvider.GetRequiredService<StreamerContext>();
-        var user = await db.Users.SingleOrDefaultAsync(a => a.PubKey == pubkey);
-
-        _cache.Set(cacheKey, user?.StreamKey);
-        return user?.StreamKey;
-    }
 }

NostrStreamer/Controllers/SRSController.cs

@@ -1,6 +1,6 @@
 using Microsoft.AspNetCore.Mvc;
 using Newtonsoft.Json;
-using NostrStreamer.Services;
+using NostrStreamer.Services.StreamManager;
 
 namespace NostrStreamer.Controllers;
 
@@ -8,14 +8,12 @@ namespace NostrStreamer.Controllers;
 public class SrsController : Controller
 {
     private readonly ILogger<SrsController> _logger;
-    private readonly Config _config;
+    private readonly StreamManagerFactory _streamManagerFactory;
-    private readonly StreamManager _streamManager;
 
-    public SrsController(ILogger<SrsController> logger, Config config, StreamManager streamManager)
+    public SrsController(ILogger<SrsController> logger, StreamManagerFactory streamManager)
     {
         _logger = logger;
-        _config = config;
+        _streamManagerFactory = streamManager;
-        _streamManager = streamManager;
     }
 
     [HttpPost]
@@ -24,8 +22,7 @@ public class SrsController : Controller
         _logger.LogInformation("OnStream: {obj}", JsonConvert.SerializeObject(req));
         try
         {
-            if (string.IsNullOrEmpty(req.Stream) || string.IsNullOrEmpty(req.App) || string.IsNullOrEmpty(req.Stream) ||
+            if (string.IsNullOrEmpty(req.Stream) || string.IsNullOrEmpty(req.App))
-                !req.App.StartsWith(_config.App, StringComparison.InvariantCultureIgnoreCase))
             {
                 return new()
                 {
@@ -33,23 +30,44 @@ public class SrsController : Controller
                 };
             }
 
+            var appSplit = req.App.Split("/");
+            var streamManager = await _streamManagerFactory.ForStream(new StreamInfo
+            {
+                App = appSplit[0],
+                Variant = appSplit.Length > 1 ? appSplit[1] : "source",
+                ClientId = req.ClientId!,
+                StreamKey = req.Stream
+            });
+
+            if (req.Action == "on_forward")
+            {
+                var urls = await streamManager.OnForward();
+                return new SrsForwardHookReply
+                {
+                    Data = new()
+                    {
+                        Urls = urls
+                    }
+                };
+            }
+            
             if (req.App.EndsWith("/source"))
             {
                 if (req.Action == "on_publish")
                 {
-                    await _streamManager.StreamStarted(req.Stream);
+                    await streamManager.StreamStarted();
                     return new();
                 }
 
                 if (req.Action == "on_unpublish")
                 {
-                    await _streamManager.StreamStopped(req.Stream);
+                    await streamManager.StreamStopped();
                     return new();
                 }
 
                 if (req.Action == "on_hls" && req.Duration.HasValue && !string.IsNullOrEmpty(req.ClientId))
                 {
-                    await _streamManager.ConsumeQuota(req.Stream, req.Duration.Value, req.ClientId);
+                    await streamManager.ConsumeQuota(req.Duration.Value);
                     return new();
                 }
             }
@@ -76,6 +94,18 @@ public class SrsHookReply
     public int Code { get; init; }
 }
 
+public class SrsForwardHookReply : SrsHookReply
+{
+    [JsonProperty("data")]
+    public SrsUrlList Data { get; init; } = null!;
+}
+
+public class SrsUrlList
+{
+    [JsonProperty("urls")]
+    public List<string> Urls { get; init; } = new();
+}
+
 public class SrsHook
 {
     [JsonProperty("action")]

NostrStreamer/Database/Configuration/IngestEndpointConfiguration.cs

@@ -0,0 +1,30 @@
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Metadata.Builders;
+
+namespace NostrStreamer.Database.Configuration;
+
+public class IngestEndpointConfiguration : IEntityTypeConfiguration<IngestEndpoint>
+{
+    public void Configure(EntityTypeBuilder<IngestEndpoint> builder)
+    {
+        builder.HasKey(a => a.Id);
+
+        builder.Property(a => a.Name)
+            .IsRequired();
+
+        builder.Property(a => a.App)
+            .IsRequired();
+
+        builder.Property(a => a.Forward)
+            .IsRequired();
+
+        builder.Property(a => a.Cost)
+            .IsRequired();
+
+        builder.Property(a => a.Capabilities)
+            .IsRequired();
+
+        builder.HasIndex(a => a.App)
+            .IsUnique();
+    }
+}

NostrStreamer/Database/Configuration/PaymentsConfiguration.cs

@@ -20,6 +20,10 @@ public class PaymentsConfiguration : IEntityTypeConfiguration<Payment>
         builder.Property(a => a.Created)
             .IsRequired();
 
+        builder.Property(a => a.Nostr);
+        builder.Property(a => a.Type)
+            .IsRequired();
+
         builder.HasOne(a => a.User)
             .WithMany(a => a.Payments)
             .HasForeignKey(a => a.PubKey);

NostrStreamer/Database/Configuration/UserConfiguration.cs

@@ -11,7 +11,6 @@ public class UserConfiguration : IEntityTypeConfiguration<User>
         builder.Property(a => a.StreamKey)
             .IsRequired();
 
-        builder.Property(a => a.Event);
         builder.Property(a => a.Balance)
             .IsRequired();
 

NostrStreamer/Database/Configuration/UserStreamConfiguration.cs

@@ -0,0 +1,35 @@
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Metadata.Builders;
+
+namespace NostrStreamer.Database.Configuration;
+
+public class UserStreamConfiguration : IEntityTypeConfiguration<UserStream>
+{
+    public void Configure(EntityTypeBuilder<UserStream> builder)
+    {
+        builder.HasKey(a => a.Id);
+        builder.Property(a => a.ClientId)
+            .IsRequired();
+
+        builder.Property(a => a.Starts)
+            .IsRequired();
+
+        builder.Property(a => a.Ends);
+
+        builder.Property(a => a.State)
+            .IsRequired();
+
+        builder.Property(a => a.Event)
+            .IsRequired();
+
+        builder.Property(a => a.Recording);
+
+        builder.HasOne(a => a.Endpoint)
+            .WithMany()
+            .HasForeignKey(a => a.EndpointId);
+
+        builder.HasOne(a => a.User)
+            .WithMany(a => a.Streams)
+            .HasForeignKey(a => a.PubKey);
+    }
+}

NostrStreamer/Database/Configuration/UserStreamGuestConfiguration.cs

@@ -0,0 +1,26 @@
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Metadata.Builders;
+
+namespace NostrStreamer.Database.Configuration;
+
+public class UserStreamGuestConfiguration : IEntityTypeConfiguration<UserStreamGuest>
+{
+    public void Configure(EntityTypeBuilder<UserStreamGuest> builder)
+    {
+        builder.HasKey(a => a.Id);
+        builder.Property(a => a.PubKey)
+            .IsRequired();
+
+        builder.Property(a => a.Sig);
+        builder.Property(a => a.Relay);
+        builder.Property(a => a.Role);
+        builder.Property(a => a.ZapSplit);
+
+        builder.HasOne(a => a.Stream)
+            .WithMany(a => a.Guests)
+            .HasForeignKey(a => a.StreamId);
+        
+        builder.HasIndex(a => new {a.StreamId, a.PubKey})
+            .IsUnique();
+    }
+}

NostrStreamer/Database/IngestEndpoint.cs

@@ -0,0 +1,28 @@
+namespace NostrStreamer.Database;
+
+public class IngestEndpoint
+{
+    public Guid Id { get; init; } = Guid.NewGuid();
+    
+    public string Name { get; init; } = null!;
+    
+    /// <summary>
+    /// Stream app name at ingest
+    /// </summary>
+    public string App { get; init; } = null!;
+
+    /// <summary>
+    /// Forward to VHost
+    /// </summary>
+    public string Forward { get; init; } = null!;
+    
+    /// <summary>
+    /// Cost/min (milli-sats)
+    /// </summary>
+    public int Cost { get; init; } = 10_000;
+
+    /// <summary>
+    /// Stream capability tags
+    /// </summary>
+    public List<string> Capabilities { get; init; } = new();
+}

NostrStreamer/Database/Payment.cs

@@ -14,4 +14,14 @@ public class Payment
     public ulong Amount { get; init; }
     
     public DateTime Created { get; init; } = DateTime.UtcNow;
+    
+    public string? Nostr { get; init; }
+    
+    public PaymentType Type { get; init; }
+}
+
+public enum PaymentType
+{
+    Topup = 0,
+    Zap = 1
 }

NostrStreamer/Database/StreamerContext.cs

@@ -6,12 +6,10 @@ public class StreamerContext : DbContext
 {
     public StreamerContext()
     {
-        
     }
 
     public StreamerContext(DbContextOptions<StreamerContext> ctx) : base(ctx)
     {
-        
     }
 
 
@@ -23,4 +21,10 @@ public class StreamerContext : DbContext
     public DbSet<User> Users => Set<User>();
 
     public DbSet<Payment> Payments => Set<Payment>();
+
+    public DbSet<UserStream> Streams => Set<UserStream>();
+
+    public DbSet<UserStreamGuest> Guests => Set<UserStreamGuest>();
+
+    public DbSet<IngestEndpoint> Endpoints => Set<IngestEndpoint>();
 }

NostrStreamer/Database/User.cs

@@ -9,11 +9,6 @@ public class User
     /// </summary>
     public string StreamKey { get; init; } = null!;
     
-    /// <summary>
-    /// Most recent nostr event published 
-    /// </summary>
-    public string? Event { get; set; }
-    
     /// <summary>
     /// Milli sats balance
     /// </summary>
@@ -50,4 +45,5 @@ public class User
     public uint Version { get; set; }
     
     public List<Payment> Payments { get; init; } = new();
+    public List<UserStream> Streams { get; init; } = new();
 }

NostrStreamer/Database/UserStream.cs

@@ -2,5 +2,38 @@ namespace NostrStreamer.Database;
 
 public class UserStream
 {
+    public Guid Id { get; init; } = Guid.NewGuid();
 
+    public string PubKey { get; init; } = null!;
+    public User User { get; init; } = null!;
+
+    public string ClientId { get; init; } = null!;
+    
+    public DateTime Starts { get; init; } = DateTime.UtcNow;
+    
+    public DateTime? Ends { get; set; }
+    
+    public UserStreamState State { get; set; }
+    
+    /// <summary>
+    /// Nostr Event for this stream
+    /// </summary>
+    public string Event { get; set; } = null!;
+    
+    /// <summary>
+    /// Recording URL of ended stream
+    /// </summary>
+    public string? Recording { get; set; }
+
+    public Guid EndpointId { get; init; }
+    public IngestEndpoint Endpoint { get; init; } = null!;
+    
+    public List<UserStreamGuest> Guests { get; init; } = new();
+}
+
+public enum UserStreamState
+{
+    Planned = 1,
+    Live = 2,
+    Ended = 3
 }

NostrStreamer/Database/UserStreamGuest.cs

@@ -0,0 +1,19 @@
+namespace NostrStreamer.Database;
+
+public class UserStreamGuest
+{
+    public Guid Id { get; init; } = Guid.NewGuid();
+
+    public Guid StreamId { get; init; }
+    public UserStream Stream { get; init; } = null!;
+    
+    public string PubKey { get; init; } = null!;
+    
+    public string? Relay { get; init; }
+    
+    public string? Role { get; init; }
+    
+    public string? Sig { get; init; }
+    
+    public decimal ZapSplit { get; init; }
+}

NostrStreamer/Extensions.cs

@@ -1,5 +1,6 @@
 using Newtonsoft.Json;
 using Nostr.Client.Json;
+using Nostr.Client.Keys;
 using Nostr.Client.Messages;
 using NostrStreamer.Database;
 
@@ -7,8 +8,91 @@ namespace NostrStreamer;
 
 public static class Extensions
 {
-    public static NostrEvent? GetNostrEvent(this User user)
+    public static NostrEvent? GetEvent(this UserStream us)
     {
-        return user.Event != default ? JsonConvert.DeserializeObject<NostrEvent>(user.Event, NostrSerializer.Settings) : null;
+        return JsonConvert.DeserializeObject<NostrEvent>(us.Event, NostrSerializer.Settings);
+    }
+
+    public static string GetPubKey(this Config cfg)
+    {
+        return NostrPrivateKey.FromBech32(cfg.PrivateKey).DerivePublicKey().Hex;
+    }
+
+    public static List<Variant> GetVariants(this IngestEndpoint ep)
+    {
+        return ep.Capabilities
+            .Where(a => a.StartsWith("variant"))
+            .Select(Variant.FromString).ToList();
+    }
+}
+
+public class Variant
+{
+    public int Width { get; init; }
+    public int Height { get; init; }
+    public int Bandwidth { get; init; }
+
+    public string SourceName => Bandwidth == -1 ? "source" : $"{Height}h";
+
+    /// <summary>
+    /// variant:{px}h:{bandwidth}
+    /// </summary>
+    /// <param name="str"></param>
+    /// <returns></returns>
+    /// <exception cref="FormatException"></exception>
+    public static Variant FromString(string str)
+    {
+        if (str.Equals("variant:source", StringComparison.InvariantCultureIgnoreCase))
+        {
+            return new()
+            {
+                Width = 0,
+                Height = 0,
+                Bandwidth = -1
+            };
+        }
+
+        var strSplit = str.Split(":");
+        if (strSplit.Length != 3 || !int.TryParse(strSplit[1][..^1], out var h) || !int.TryParse(strSplit[2], out var b))
+        {
+            throw new FormatException();
+        }
+
+        return new()
+        {
+            Height = h,
+            Width = (int)Math.Ceiling(h / 9m * 16m),
+            Bandwidth = b
+        };
+    }
+
+    public override string ToString()
+    {
+        if (Bandwidth == -1)
+        {
+            return "variant:source";
+        }
+
+        return $"variant:{SourceName}:{Bandwidth}";
+    }
+
+    public string ToResolutionArg()
+    {
+        if (Bandwidth == -1)
+        {
+            return string.Empty;
+        }
+
+        return $"RESOLUTION={Width}x{Height}";
+    }
+
+    public string ToBandwidthArg()
+    {
+        if (Bandwidth == -1)
+        {
+            return string.Empty;
+        }
+
+        return $"BANDWIDTH={Bandwidth * 1000}";
     }
 }

NostrStreamer/Migrations/20230725105922_V2.Designer.cs

@@ -0,0 +1,218 @@
+// <auto-generated />
+using System;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Infrastructure;
+using Microsoft.EntityFrameworkCore.Migrations;
+using Microsoft.EntityFrameworkCore.Storage.ValueConversion;
+using NostrStreamer.Database;
+using Npgsql.EntityFrameworkCore.PostgreSQL.Metadata;
+
+#nullable disable
+
+namespace NostrStreamer.Migrations
+{
+    [DbContext(typeof(StreamerContext))]
+    [Migration("20230725105922_V2")]
+    partial class V2
+    {
+        /// <inheritdoc />
+        protected override void BuildTargetModel(ModelBuilder modelBuilder)
+        {
+#pragma warning disable 612, 618
+            modelBuilder
+                .HasAnnotation("ProductVersion", "7.0.8")
+                .HasAnnotation("Relational:MaxIdentifierLength", 63);
+
+            NpgsqlModelBuilderExtensions.UseIdentityByDefaultColumns(modelBuilder);
+
+            modelBuilder.Entity("NostrStreamer.Database.Payment", b =>
+                {
+                    b.Property<string>("PaymentHash")
+                        .HasColumnType("text");
+
+                    b.Property<decimal>("Amount")
+                        .HasColumnType("numeric(20,0)");
+
+                    b.Property<DateTime>("Created")
+                        .HasColumnType("timestamp with time zone");
+
+                    b.Property<string>("Invoice")
+                        .IsRequired()
+                        .HasColumnType("text");
+
+                    b.Property<bool>("IsPaid")
+                        .HasColumnType("boolean");
+
+                    b.Property<string>("Nostr")
+                        .HasColumnType("text");
+
+                    b.Property<string>("PubKey")
+                        .IsRequired()
+                        .HasColumnType("text");
+
+                    b.Property<int>("Type")
+                        .HasColumnType("integer");
+
+                    b.HasKey("PaymentHash");
+
+                    b.HasIndex("PubKey");
+
+                    b.ToTable("Payments");
+                });
+
+            modelBuilder.Entity("NostrStreamer.Database.User", b =>
+                {
+                    b.Property<string>("PubKey")
+                        .HasColumnType("text");
+
+                    b.Property<long>("Balance")
+                        .HasColumnType("bigint");
+
+                    b.Property<string>("ContentWarning")
+                        .HasColumnType("text");
+
+                    b.Property<string>("Image")
+                        .HasColumnType("text");
+
+                    b.Property<string>("StreamKey")
+                        .IsRequired()
+                        .HasColumnType("text");
+
+                    b.Property<string>("Summary")
+                        .HasColumnType("text");
+
+                    b.Property<string>("Tags")
+                        .HasColumnType("text");
+
+                    b.Property<string>("Title")
+                        .HasColumnType("text");
+
+                    b.Property<uint>("Version")
+                        .IsConcurrencyToken()
+                        .ValueGeneratedOnAddOrUpdate()
+                        .HasColumnType("xid")
+                        .HasColumnName("xmin");
+
+                    b.HasKey("PubKey");
+
+                    b.ToTable("Users");
+                });
+
+            modelBuilder.Entity("NostrStreamer.Database.UserStream", b =>
+                {
+                    b.Property<Guid>("Id")
+                        .ValueGeneratedOnAdd()
+                        .HasColumnType("uuid");
+
+                    b.Property<string>("ClientId")
+                        .IsRequired()
+                        .HasColumnType("text");
+
+                    b.Property<DateTime?>("Ends")
+                        .HasColumnType("timestamp with time zone");
+
+                    b.Property<string>("Event")
+                        .IsRequired()
+                        .HasColumnType("text");
+
+                    b.Property<string>("PubKey")
+                        .IsRequired()
+                        .HasColumnType("text");
+
+                    b.Property<string>("Recording")
+                        .HasColumnType("text");
+
+                    b.Property<DateTime>("Starts")
+                        .HasColumnType("timestamp with time zone");
+
+                    b.Property<int>("State")
+                        .HasColumnType("integer");
+
+                    b.HasKey("Id");
+
+                    b.HasIndex("PubKey");
+
+                    b.ToTable("Streams");
+                });
+
+            modelBuilder.Entity("NostrStreamer.Database.UserStreamGuest", b =>
+                {
+                    b.Property<Guid>("Id")
+                        .ValueGeneratedOnAdd()
+                        .HasColumnType("uuid");
+
+                    b.Property<string>("PubKey")
+                        .IsRequired()
+                        .HasColumnType("text");
+
+                    b.Property<string>("Relay")
+                        .HasColumnType("text");
+
+                    b.Property<string>("Role")
+                        .HasColumnType("text");
+
+                    b.Property<string>("Sig")
+                        .HasColumnType("text");
+
+                    b.Property<Guid>("StreamId")
+                        .HasColumnType("uuid");
+
+                    b.Property<decimal>("ZapSplit")
+                        .HasColumnType("numeric");
+
+                    b.HasKey("Id");
+
+                    b.HasIndex("StreamId", "PubKey")
+                        .IsUnique();
+
+                    b.ToTable("Guests");
+                });
+
+            modelBuilder.Entity("NostrStreamer.Database.Payment", b =>
+                {
+                    b.HasOne("NostrStreamer.Database.User", "User")
+                        .WithMany("Payments")
+                        .HasForeignKey("PubKey")
+                        .OnDelete(DeleteBehavior.Cascade)
+                        .IsRequired();
+
+                    b.Navigation("User");
+                });
+
+            modelBuilder.Entity("NostrStreamer.Database.UserStream", b =>
+                {
+                    b.HasOne("NostrStreamer.Database.User", "User")
+                        .WithMany("Streams")
+                        .HasForeignKey("PubKey")
+                        .OnDelete(DeleteBehavior.Cascade)
+                        .IsRequired();
+
+                    b.Navigation("User");
+                });
+
+            modelBuilder.Entity("NostrStreamer.Database.UserStreamGuest", b =>
+                {
+                    b.HasOne("NostrStreamer.Database.UserStream", "Stream")
+                        .WithMany("Guests")
+                        .HasForeignKey("StreamId")
+                        .OnDelete(DeleteBehavior.Cascade)
+                        .IsRequired();
+
+                    b.Navigation("Stream");
+                });
+
+            modelBuilder.Entity("NostrStreamer.Database.User", b =>
+                {
+                    b.Navigation("Payments");
+
+                    b.Navigation("Streams");
+                });
+
+            modelBuilder.Entity("NostrStreamer.Database.UserStream", b =>
+                {
+                    b.Navigation("Guests");
+                });
+#pragma warning restore 612, 618
+        }
+    }
+}

NostrStreamer/Migrations/20230725105922_V2.cs

@@ -0,0 +1,114 @@
+using System;
+using Microsoft.EntityFrameworkCore.Migrations;
+
+#nullable disable
+
+namespace NostrStreamer.Migrations
+{
+    /// <inheritdoc />
+    public partial class V2 : Migration
+    {
+        /// <inheritdoc />
+        protected override void Up(MigrationBuilder migrationBuilder)
+        {
+            migrationBuilder.DropColumn(
+                name: "Event",
+                table: "Users");
+
+            migrationBuilder.AddColumn<string>(
+                name: "Nostr",
+                table: "Payments",
+                type: "text",
+                nullable: true);
+
+            migrationBuilder.AddColumn<int>(
+                name: "Type",
+                table: "Payments",
+                type: "integer",
+                nullable: false,
+                defaultValue: 0);
+
+            migrationBuilder.CreateTable(
+                name: "Streams",
+                columns: table => new
+                {
+                    Id = table.Column<Guid>(type: "uuid", nullable: false),
+                    PubKey = table.Column<string>(type: "text", nullable: false),
+                    ClientId = table.Column<string>(type: "text", nullable: false),
+                    Starts = table.Column<DateTime>(type: "timestamp with time zone", nullable: false),
+                    Ends = table.Column<DateTime>(type: "timestamp with time zone", nullable: true),
+                    State = table.Column<int>(type: "integer", nullable: false),
+                    Event = table.Column<string>(type: "text", nullable: false),
+                    Recording = table.Column<string>(type: "text", nullable: true)
+                },
+                constraints: table =>
+                {
+                    table.PrimaryKey("PK_Streams", x => x.Id);
+                    table.ForeignKey(
+                        name: "FK_Streams_Users_PubKey",
+                        column: x => x.PubKey,
+                        principalTable: "Users",
+                        principalColumn: "PubKey",
+                        onDelete: ReferentialAction.Cascade);
+                });
+
+            migrationBuilder.CreateTable(
+                name: "Guests",
+                columns: table => new
+                {
+                    Id = table.Column<Guid>(type: "uuid", nullable: false),
+                    StreamId = table.Column<Guid>(type: "uuid", nullable: false),
+                    PubKey = table.Column<string>(type: "text", nullable: false),
+                    Relay = table.Column<string>(type: "text", nullable: true),
+                    Role = table.Column<string>(type: "text", nullable: true),
+                    Sig = table.Column<string>(type: "text", nullable: true),
+                    ZapSplit = table.Column<decimal>(type: "numeric", nullable: false)
+                },
+                constraints: table =>
+                {
+                    table.PrimaryKey("PK_Guests", x => x.Id);
+                    table.ForeignKey(
+                        name: "FK_Guests_Streams_StreamId",
+                        column: x => x.StreamId,
+                        principalTable: "Streams",
+                        principalColumn: "Id",
+                        onDelete: ReferentialAction.Cascade);
+                });
+
+            migrationBuilder.CreateIndex(
+                name: "IX_Guests_StreamId_PubKey",
+                table: "Guests",
+                columns: new[] { "StreamId", "PubKey" },
+                unique: true);
+
+            migrationBuilder.CreateIndex(
+                name: "IX_Streams_PubKey",
+                table: "Streams",
+                column: "PubKey");
+        }
+
+        /// <inheritdoc />
+        protected override void Down(MigrationBuilder migrationBuilder)
+        {
+            migrationBuilder.DropTable(
+                name: "Guests");
+
+            migrationBuilder.DropTable(
+                name: "Streams");
+
+            migrationBuilder.DropColumn(
+                name: "Nostr",
+                table: "Payments");
+
+            migrationBuilder.DropColumn(
+                name: "Type",
+                table: "Payments");
+
+            migrationBuilder.AddColumn<string>(
+                name: "Event",
+                table: "Users",
+                type: "text",
+                nullable: true);
+        }
+    }
+}

NostrStreamer/Migrations/20230725123250_Endpoints.Designer.cs

@@ -0,0 +1,265 @@
+// <auto-generated />
+using System;
+using System.Collections.Generic;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Infrastructure;
+using Microsoft.EntityFrameworkCore.Migrations;
+using Microsoft.EntityFrameworkCore.Storage.ValueConversion;
+using NostrStreamer.Database;
+using Npgsql.EntityFrameworkCore.PostgreSQL.Metadata;
+
+#nullable disable
+
+namespace NostrStreamer.Migrations
+{
+    [DbContext(typeof(StreamerContext))]
+    [Migration("20230725123250_Endpoints")]
+    partial class Endpoints
+    {
+        /// <inheritdoc />
+        protected override void BuildTargetModel(ModelBuilder modelBuilder)
+        {
+#pragma warning disable 612, 618
+            modelBuilder
+                .HasAnnotation("ProductVersion", "7.0.8")
+                .HasAnnotation("Relational:MaxIdentifierLength", 63);
+
+            NpgsqlModelBuilderExtensions.UseIdentityByDefaultColumns(modelBuilder);
+
+            modelBuilder.Entity("NostrStreamer.Database.IngestEndpoint", b =>
+                {
+                    b.Property<Guid>("Id")
+                        .ValueGeneratedOnAdd()
+                        .HasColumnType("uuid");
+
+                    b.Property<string>("App")
+                        .IsRequired()
+                        .HasColumnType("text");
+
+                    b.Property<List<string>>("Capabilities")
+                        .IsRequired()
+                        .HasColumnType("text[]");
+
+                    b.Property<int>("Cost")
+                        .HasColumnType("integer");
+
+                    b.Property<string>("Forward")
+                        .IsRequired()
+                        .HasColumnType("text");
+
+                    b.Property<string>("Name")
+                        .IsRequired()
+                        .HasColumnType("text");
+
+                    b.HasKey("Id");
+
+                    b.HasIndex("App")
+                        .IsUnique();
+
+                    b.ToTable("Endpoints");
+                });
+
+            modelBuilder.Entity("NostrStreamer.Database.Payment", b =>
+                {
+                    b.Property<string>("PaymentHash")
+                        .HasColumnType("text");
+
+                    b.Property<decimal>("Amount")
+                        .HasColumnType("numeric(20,0)");
+
+                    b.Property<DateTime>("Created")
+                        .HasColumnType("timestamp with time zone");
+
+                    b.Property<string>("Invoice")
+                        .IsRequired()
+                        .HasColumnType("text");
+
+                    b.Property<bool>("IsPaid")
+                        .HasColumnType("boolean");
+
+                    b.Property<string>("Nostr")
+                        .HasColumnType("text");
+
+                    b.Property<string>("PubKey")
+                        .IsRequired()
+                        .HasColumnType("text");
+
+                    b.Property<int>("Type")
+                        .HasColumnType("integer");
+
+                    b.HasKey("PaymentHash");
+
+                    b.HasIndex("PubKey");
+
+                    b.ToTable("Payments");
+                });
+
+            modelBuilder.Entity("NostrStreamer.Database.User", b =>
+                {
+                    b.Property<string>("PubKey")
+                        .HasColumnType("text");
+
+                    b.Property<long>("Balance")
+                        .HasColumnType("bigint");
+
+                    b.Property<string>("ContentWarning")
+                        .HasColumnType("text");
+
+                    b.Property<string>("Image")
+                        .HasColumnType("text");
+
+                    b.Property<string>("StreamKey")
+                        .IsRequired()
+                        .HasColumnType("text");
+
+                    b.Property<string>("Summary")
+                        .HasColumnType("text");
+
+                    b.Property<string>("Tags")
+                        .HasColumnType("text");
+
+                    b.Property<string>("Title")
+                        .HasColumnType("text");
+
+                    b.Property<uint>("Version")
+                        .IsConcurrencyToken()
+                        .ValueGeneratedOnAddOrUpdate()
+                        .HasColumnType("xid")
+                        .HasColumnName("xmin");
+
+                    b.HasKey("PubKey");
+
+                    b.ToTable("Users");
+                });
+
+            modelBuilder.Entity("NostrStreamer.Database.UserStream", b =>
+                {
+                    b.Property<Guid>("Id")
+                        .ValueGeneratedOnAdd()
+                        .HasColumnType("uuid");
+
+                    b.Property<string>("ClientId")
+                        .IsRequired()
+                        .HasColumnType("text");
+
+                    b.Property<Guid>("EndpointId")
+                        .HasColumnType("uuid");
+
+                    b.Property<DateTime?>("Ends")
+                        .HasColumnType("timestamp with time zone");
+
+                    b.Property<string>("Event")
+                        .IsRequired()
+                        .HasColumnType("text");
+
+                    b.Property<string>("PubKey")
+                        .IsRequired()
+                        .HasColumnType("text");
+
+                    b.Property<string>("Recording")
+                        .HasColumnType("text");
+
+                    b.Property<DateTime>("Starts")
+                        .HasColumnType("timestamp with time zone");
+
+                    b.Property<int>("State")
+                        .HasColumnType("integer");
+
+                    b.HasKey("Id");
+
+                    b.HasIndex("EndpointId");
+
+                    b.HasIndex("PubKey");
+
+                    b.ToTable("Streams");
+                });
+
+            modelBuilder.Entity("NostrStreamer.Database.UserStreamGuest", b =>
+                {
+                    b.Property<Guid>("Id")
+                        .ValueGeneratedOnAdd()
+                        .HasColumnType("uuid");
+
+                    b.Property<string>("PubKey")
+                        .IsRequired()
+                        .HasColumnType("text");
+
+                    b.Property<string>("Relay")
+                        .HasColumnType("text");
+
+                    b.Property<string>("Role")
+                        .HasColumnType("text");
+
+                    b.Property<string>("Sig")
+                        .HasColumnType("text");
+
+                    b.Property<Guid>("StreamId")
+                        .HasColumnType("uuid");
+
+                    b.Property<decimal>("ZapSplit")
+                        .HasColumnType("numeric");
+
+                    b.HasKey("Id");
+
+                    b.HasIndex("StreamId", "PubKey")
+                        .IsUnique();
+
+                    b.ToTable("Guests");
+                });
+
+            modelBuilder.Entity("NostrStreamer.Database.Payment", b =>
+                {
+                    b.HasOne("NostrStreamer.Database.User", "User")
+                        .WithMany("Payments")
+                        .HasForeignKey("PubKey")
+                        .OnDelete(DeleteBehavior.Cascade)
+                        .IsRequired();
+
+                    b.Navigation("User");
+                });
+
+            modelBuilder.Entity("NostrStreamer.Database.UserStream", b =>
+                {
+                    b.HasOne("NostrStreamer.Database.IngestEndpoint", "Endpoint")
+                        .WithMany()
+                        .HasForeignKey("EndpointId")
+                        .OnDelete(DeleteBehavior.Cascade)
+                        .IsRequired();
+
+                    b.HasOne("NostrStreamer.Database.User", "User")
+                        .WithMany("Streams")
+                        .HasForeignKey("PubKey")
+                        .OnDelete(DeleteBehavior.Cascade)
+                        .IsRequired();
+
+                    b.Navigation("Endpoint");
+
+                    b.Navigation("User");
+                });
+
+            modelBuilder.Entity("NostrStreamer.Database.UserStreamGuest", b =>
+                {
+                    b.HasOne("NostrStreamer.Database.UserStream", "Stream")
+                        .WithMany("Guests")
+                        .HasForeignKey("StreamId")
+                        .OnDelete(DeleteBehavior.Cascade)
+                        .IsRequired();
+
+                    b.Navigation("Stream");
+                });
+
+            modelBuilder.Entity("NostrStreamer.Database.User", b =>
+                {
+                    b.Navigation("Payments");
+
+                    b.Navigation("Streams");
+                });
+
+            modelBuilder.Entity("NostrStreamer.Database.UserStream", b =>
+                {
+                    b.Navigation("Guests");
+                });
+#pragma warning restore 612, 618
+        }
+    }
+}

NostrStreamer/Migrations/20230725123250_Endpoints.cs

@@ -0,0 +1,77 @@
+using System;
+using System.Collections.Generic;
+using Microsoft.EntityFrameworkCore.Migrations;
+
+#nullable disable
+
+namespace NostrStreamer.Migrations
+{
+    /// <inheritdoc />
+    public partial class Endpoints : Migration
+    {
+        /// <inheritdoc />
+        protected override void Up(MigrationBuilder migrationBuilder)
+        {
+            migrationBuilder.AddColumn<Guid>(
+                name: "EndpointId",
+                table: "Streams",
+                type: "uuid",
+                nullable: false,
+                defaultValue: new Guid("00000000-0000-0000-0000-000000000000"));
+
+            migrationBuilder.CreateTable(
+                name: "Endpoints",
+                columns: table => new
+                {
+                    Id = table.Column<Guid>(type: "uuid", nullable: false),
+                    Name = table.Column<string>(type: "text", nullable: false),
+                    App = table.Column<string>(type: "text", nullable: false),
+                    Forward = table.Column<string>(type: "text", nullable: false),
+                    Cost = table.Column<int>(type: "integer", nullable: false),
+                    Capabilities = table.Column<List<string>>(type: "text[]", nullable: false)
+                },
+                constraints: table =>
+                {
+                    table.PrimaryKey("PK_Endpoints", x => x.Id);
+                });
+
+            migrationBuilder.CreateIndex(
+                name: "IX_Streams_EndpointId",
+                table: "Streams",
+                column: "EndpointId");
+
+            migrationBuilder.CreateIndex(
+                name: "IX_Endpoints_App",
+                table: "Endpoints",
+                column: "App",
+                unique: true);
+
+            migrationBuilder.AddForeignKey(
+                name: "FK_Streams_Endpoints_EndpointId",
+                table: "Streams",
+                column: "EndpointId",
+                principalTable: "Endpoints",
+                principalColumn: "Id",
+                onDelete: ReferentialAction.Cascade);
+        }
+
+        /// <inheritdoc />
+        protected override void Down(MigrationBuilder migrationBuilder)
+        {
+            migrationBuilder.DropForeignKey(
+                name: "FK_Streams_Endpoints_EndpointId",
+                table: "Streams");
+
+            migrationBuilder.DropTable(
+                name: "Endpoints");
+
+            migrationBuilder.DropIndex(
+                name: "IX_Streams_EndpointId",
+                table: "Streams");
+
+            migrationBuilder.DropColumn(
+                name: "EndpointId",
+                table: "Streams");
+        }
+    }
+}

NostrStreamer/Migrations/StreamerContextModelSnapshot.cs

@@ -1,5 +1,6 @@
 // <auto-generated />
 using System;
+using System.Collections.Generic;
 using Microsoft.EntityFrameworkCore;
 using Microsoft.EntityFrameworkCore.Infrastructure;
 using Microsoft.EntityFrameworkCore.Storage.ValueConversion;
@@ -22,6 +23,39 @@ namespace NostrStreamer.Migrations
 
             NpgsqlModelBuilderExtensions.UseIdentityByDefaultColumns(modelBuilder);
 
+            modelBuilder.Entity("NostrStreamer.Database.IngestEndpoint", b =>
+                {
+                    b.Property<Guid>("Id")
+                        .ValueGeneratedOnAdd()
+                        .HasColumnType("uuid");
+
+                    b.Property<string>("App")
+                        .IsRequired()
+                        .HasColumnType("text");
+
+                    b.Property<List<string>>("Capabilities")
+                        .IsRequired()
+                        .HasColumnType("text[]");
+
+                    b.Property<double>("Cost")
+                        .HasColumnType("integer");
+
+                    b.Property<string>("Forward")
+                        .IsRequired()
+                        .HasColumnType("text");
+
+                    b.Property<string>("Name")
+                        .IsRequired()
+                        .HasColumnType("text");
+
+                    b.HasKey("Id");
+
+                    b.HasIndex("App")
+                        .IsUnique();
+
+                    b.ToTable("Endpoints");
+                });
+
             modelBuilder.Entity("NostrStreamer.Database.Payment", b =>
                 {
                     b.Property<string>("PaymentHash")
@@ -40,10 +74,16 @@ namespace NostrStreamer.Migrations
                     b.Property<bool>("IsPaid")
                         .HasColumnType("boolean");
 
+                    b.Property<string>("Nostr")
+                        .HasColumnType("text");
+
                     b.Property<string>("PubKey")
                         .IsRequired()
                         .HasColumnType("text");
 
+                    b.Property<int>("Type")
+                        .HasColumnType("integer");
+
                     b.HasKey("PaymentHash");
 
                     b.HasIndex("PubKey");
@@ -62,9 +102,6 @@ namespace NostrStreamer.Migrations
                     b.Property<string>("ContentWarning")
                         .HasColumnType("text");
 
-                    b.Property<string>("Event")
-                        .HasColumnType("text");
-
                     b.Property<string>("Image")
                         .HasColumnType("text");
 
@@ -92,6 +129,81 @@ namespace NostrStreamer.Migrations
                     b.ToTable("Users");
                 });
 
+            modelBuilder.Entity("NostrStreamer.Database.UserStream", b =>
+                {
+                    b.Property<Guid>("Id")
+                        .ValueGeneratedOnAdd()
+                        .HasColumnType("uuid");
+
+                    b.Property<string>("ClientId")
+                        .IsRequired()
+                        .HasColumnType("text");
+
+                    b.Property<Guid>("EndpointId")
+                        .HasColumnType("uuid");
+
+                    b.Property<DateTime?>("Ends")
+                        .HasColumnType("timestamp with time zone");
+
+                    b.Property<string>("Event")
+                        .IsRequired()
+                        .HasColumnType("text");
+
+                    b.Property<string>("PubKey")
+                        .IsRequired()
+                        .HasColumnType("text");
+
+                    b.Property<string>("Recording")
+                        .HasColumnType("text");
+
+                    b.Property<DateTime>("Starts")
+                        .HasColumnType("timestamp with time zone");
+
+                    b.Property<int>("State")
+                        .HasColumnType("integer");
+
+                    b.HasKey("Id");
+
+                    b.HasIndex("EndpointId");
+
+                    b.HasIndex("PubKey");
+
+                    b.ToTable("Streams");
+                });
+
+            modelBuilder.Entity("NostrStreamer.Database.UserStreamGuest", b =>
+                {
+                    b.Property<Guid>("Id")
+                        .ValueGeneratedOnAdd()
+                        .HasColumnType("uuid");
+
+                    b.Property<string>("PubKey")
+                        .IsRequired()
+                        .HasColumnType("text");
+
+                    b.Property<string>("Relay")
+                        .HasColumnType("text");
+
+                    b.Property<string>("Role")
+                        .HasColumnType("text");
+
+                    b.Property<string>("Sig")
+                        .HasColumnType("text");
+
+                    b.Property<Guid>("StreamId")
+                        .HasColumnType("uuid");
+
+                    b.Property<decimal>("ZapSplit")
+                        .HasColumnType("numeric");
+
+                    b.HasKey("Id");
+
+                    b.HasIndex("StreamId", "PubKey")
+                        .IsUnique();
+
+                    b.ToTable("Guests");
+                });
+
             modelBuilder.Entity("NostrStreamer.Database.Payment", b =>
                 {
                     b.HasOne("NostrStreamer.Database.User", "User")
@@ -103,9 +215,46 @@ namespace NostrStreamer.Migrations
                     b.Navigation("User");
                 });
 
+            modelBuilder.Entity("NostrStreamer.Database.UserStream", b =>
+                {
+                    b.HasOne("NostrStreamer.Database.IngestEndpoint", "Endpoint")
+                        .WithMany()
+                        .HasForeignKey("EndpointId")
+                        .OnDelete(DeleteBehavior.Cascade)
+                        .IsRequired();
+
+                    b.HasOne("NostrStreamer.Database.User", "User")
+                        .WithMany("Streams")
+                        .HasForeignKey("PubKey")
+                        .OnDelete(DeleteBehavior.Cascade)
+                        .IsRequired();
+
+                    b.Navigation("Endpoint");
+
+                    b.Navigation("User");
+                });
+
+            modelBuilder.Entity("NostrStreamer.Database.UserStreamGuest", b =>
+                {
+                    b.HasOne("NostrStreamer.Database.UserStream", "Stream")
+                        .WithMany("Guests")
+                        .HasForeignKey("StreamId")
+                        .OnDelete(DeleteBehavior.Cascade)
+                        .IsRequired();
+
+                    b.Navigation("Stream");
+                });
+
             modelBuilder.Entity("NostrStreamer.Database.User", b =>
                 {
                     b.Navigation("Payments");
+
+                    b.Navigation("Streams");
+                });
+
+            modelBuilder.Entity("NostrStreamer.Database.UserStream", b =>
+                {
+                    b.Navigation("Guests");
                 });
 #pragma warning restore 612, 618
         }

NostrStreamer/NostrStreamer.csproj

@@ -20,8 +20,11 @@
       <Content Include="..\docker-compose.yaml">
         <Link>docker-compose.yaml</Link>
       </Content>
-      <Content Include="..\docker\srs.conf">
+      <Content Include="..\docker\srs-edge.conf">
-        <Link>srs.conf</Link>
+        <Link>srs-edge.conf</Link>
+      </Content>
+      <Content Include="..\docker\srs-origin.conf">
+        <Link>srs-origin.conf</Link>
       </Content>
     </ItemGroup>
 
@@ -32,6 +35,7 @@
         <PrivateAssets>all</PrivateAssets>
         <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
       </PackageReference>
+      <PackageReference Include="LNURL" Version="0.0.30" />
       <PackageReference Include="Microsoft.AspNetCore.Mvc.NewtonsoftJson" Version="6.0.19" />
       <PackageReference Include="Microsoft.EntityFrameworkCore" Version="7.0.8" />
       <PackageReference Include="Microsoft.EntityFrameworkCore.Design" Version="7.0.8">
@@ -41,4 +45,8 @@
       <PackageReference Include="Nostr.Client" Version="1.4.2" />
       <PackageReference Include="Npgsql.EntityFrameworkCore.PostgreSQL" Version="7.0.4" />
     </ItemGroup>
+
+    <ItemGroup>
+      <Folder Include="Services\Abstract\" />
+    </ItemGroup>
 </Project>

NostrStreamer/Program.cs

@@ -5,6 +5,7 @@ using Microsoft.EntityFrameworkCore;
 using Nostr.Client.Client;
 using NostrStreamer.Database;
 using NostrStreamer.Services;
+using NostrStreamer.Services.StreamManager;
 
 namespace NostrStreamer;
 
@@ -47,11 +48,13 @@ internal static class Program
         services.AddHostedService<NostrListenerLifetime>();
         
         // streaming services
-        services.AddTransient<StreamManager>();
         services.AddTransient<SrsApi>();
         services.AddHostedService<BackgroundStreamManager>();
         services.AddSingleton<ViewCounter>();
         services.AddHostedService<ViewCounterDecay>();
+        services.AddTransient<StreamEventBuilder>();
+        services.AddTransient<StreamManagerFactory>();
+        services.AddTransient<UserService>();
         
         // lnd services
         services.AddSingleton<LndNode>();

NostrStreamer/Services/BackgroundStreamManager.cs

@@ -1,3 +1,7 @@
+using Microsoft.EntityFrameworkCore;
+using NostrStreamer.Database;
+using NostrStreamer.Services.StreamManager;
+
 namespace NostrStreamer.Services;
 
 public class BackgroundStreamManager : BackgroundService
@@ -19,15 +23,19 @@ public class BackgroundStreamManager : BackgroundService
             {
                 using var scope = _scopeFactory.CreateScope();
 
-                var streamManager = scope.ServiceProvider.GetRequiredService<StreamManager>();
+                var streamManager = scope.ServiceProvider.GetRequiredService<StreamManagerFactory>();
-                var srsApi = scope.ServiceProvider.GetRequiredService<SrsApi>();
+                var db = scope.ServiceProvider.GetRequiredService<StreamerContext>();
-                var viewCounter = scope.ServiceProvider.GetRequiredService<ViewCounter>();
 
-                var streams = await srsApi.ListStreams();
+                var liveStreams = await db.Streams
-                foreach (var stream in streams.GroupBy(a => a.Name))
+                    .AsNoTracking()
+                    .Where(a => a.State == UserStreamState.Live)
+                    .Select(a => a.Id)
+                    .ToListAsync(cancellationToken: stoppingToken);
+
+                foreach (var id in liveStreams)
                 {
-                    var viewers = viewCounter.Current(stream.Key);
+                    var manager = await streamManager.ForStream(id);
-                    await streamManager.UpdateViewers(stream.Key, viewers);
+                    await manager.UpdateViewers();
                 }
             }
             catch (Exception ex)

NostrStreamer/Services/SrsApi.cs

@@ -18,6 +18,12 @@ public class SrsApi
         return rsp!.Streams;
     }
 
+    public async Task<Stream> GetStream(string id)
+    {
+        var rsp = await _client.GetFromJsonAsync<Stream>($"/api/v1/streams/{id}");
+        return rsp!;
+    }
+
     public async Task<List<Client>> ListClients()
     {
         var rsp = await _client.GetFromJsonAsync<ListClientsResponse>("/api/v1/clients/?count=10000");

NostrStreamer/Services/StreamEventBuilder.cs

@@ -0,0 +1,97 @@
+using System.ComponentModel;
+using Nostr.Client.Client;
+using Nostr.Client.Keys;
+using Nostr.Client.Messages;
+using Nostr.Client.Requests;
+using NostrStreamer.Database;
+
+namespace NostrStreamer.Services;
+
+public class StreamEventBuilder
+{
+    private const NostrKind StreamEventKind = (NostrKind)30_311;
+    private const NostrKind StreamChatKind = (NostrKind)1311;
+    private readonly Config _config;
+    private readonly ViewCounter _viewCounter;
+    private readonly INostrClient _nostrClient;
+
+    public StreamEventBuilder(Config config, ViewCounter viewCounter, INostrClient nostrClient)
+    {
+        _config = config;
+        _viewCounter = viewCounter;
+        _nostrClient = nostrClient;
+    }
+
+    public NostrEvent CreateStreamEvent(User user, UserStream stream)
+    {
+        var status = stream.State switch
+        {
+            UserStreamState.Planned => "planned",
+            UserStreamState.Live => "live",
+            UserStreamState.Ended => "ended",
+            _ => throw new InvalidEnumArgumentException()
+        };
+
+        var tags = new List<NostrEventTag>
+        {
+            new("d", stream.Id.ToString()),
+            new("title", user.Title ?? ""),
+            new("summary", user.Summary ?? ""),
+            new("streaming", new Uri(_config.DataHost, $"{user.PubKey}.m3u8").ToString()),
+            new("image", user.Image ?? new Uri(_config.DataHost, $"{user.PubKey}.png").ToString()),
+            new("status", status),
+            new("p", user.PubKey, "", "host"),
+            new("relays", _config.Relays),
+        };
+
+        if (status == "live")
+        {
+            var viewers = _viewCounter.Current(stream.Id);
+            var starts = new DateTimeOffset(stream.Starts).ToUnixTimeSeconds();
+            tags.Add(new("starts", starts.ToString()));
+            tags.Add(new("current_participants", viewers.ToString()));
+
+            if (!string.IsNullOrEmpty(user.ContentWarning))
+            {
+                tags.Add(new("content-warning", user.ContentWarning));
+            }
+        }
+
+        foreach (var tag in !string.IsNullOrEmpty(user.Tags) ?
+                     user.Tags.Split(",", StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries) : Array.Empty<string>())
+        {
+            tags.Add(new("t", tag));
+        }
+
+        var ev = new NostrEvent
+        {
+            Kind = StreamEventKind,
+            Content = "",
+            CreatedAt = DateTime.Now,
+            Tags = new NostrEventTags(tags)
+        };
+
+        return ev.Sign(NostrPrivateKey.FromBech32(_config.PrivateKey));
+    }
+    
+    public NostrEvent CreateStreamChat(UserStream stream, string message)
+    {
+        var pk = NostrPrivateKey.FromBech32(_config.PrivateKey);
+        var ev = new NostrEvent
+        {
+            Kind = StreamChatKind,
+            Content = message,
+            CreatedAt = DateTime.Now,
+            Tags = new NostrEventTags(
+                new NostrEventTag("a", $"{StreamEventKind}:{pk.DerivePublicKey().Hex}:{stream.Id}")
+            )
+        };
+
+        return ev.Sign(pk);
+    }
+
+    public void BroadcastEvent(NostrEvent ev)
+    {
+        _nostrClient.Send(new NostrEventRequest(ev));
+    }
+}

NostrStreamer/Services/StreamManager.cs

@@ -1,205 +0,0 @@
-using Microsoft.EntityFrameworkCore;
-using Newtonsoft.Json;
-using Nostr.Client.Client;
-using Nostr.Client.Json;
-using Nostr.Client.Keys;
-using Nostr.Client.Messages;
-using Nostr.Client.Requests;
-using NostrStreamer.Database;
-
-namespace NostrStreamer.Services;
-
-public class StreamManager
-{
-    private const NostrKind StreamEventKind = (NostrKind)30_311;
-    private const NostrKind StreamChatKind = (NostrKind)1311;
-
-    private readonly ILogger<StreamManager> _logger;
-    private readonly StreamerContext _db;
-    private readonly Config _config;
-    private readonly INostrClient _nostr;
-    private readonly SrsApi _srsApi;
-
-    public StreamManager(ILogger<StreamManager> logger, StreamerContext db, Config config, INostrClient nostr, SrsApi srsApi)
-    {
-        _logger = logger;
-        _db = db;
-        _config = config;
-        _nostr = nostr;
-        _srsApi = srsApi;
-    }
-
-    public async Task StreamStarted(string streamKey)
-    {
-        var user = await GetUserFromStreamKey(streamKey);
-        if (user == default) throw new Exception("No stream key found");
-
-        _logger.LogInformation("Stream started for: {pubkey}", user.PubKey);
-
-        if (user.Balance <= 0)
-        {
-            throw new Exception("User balance empty");
-        }
-
-        var ev = CreateStreamEvent(user, "live");
-        await PublishEvent(user, ev);
-    }
-
-    public async Task StreamStopped(string streamKey)
-    {
-        var user = await GetUserFromStreamKey(streamKey);
-        if (user == default) throw new Exception("No stream key found");
-
-        _logger.LogInformation("Stream stopped for: {pubkey}", user.PubKey);
-
-        var ev = CreateStreamEvent(user, "ended");
-        await PublishEvent(user, ev);
-    }
-
-    public async Task ConsumeQuota(string streamKey, double duration, string clientId)
-    {
-        var user = await GetUserFromStreamKey(streamKey);
-        if (user == default) throw new Exception("No stream key found");
-
-        const long balanceAlertThreshold = 500_000;
-        var cost = (long)Math.Ceiling(_config.Cost * (duration / 60d));
-        if (cost > 0)
-        {
-            await _db.Users
-                .Where(a => a.PubKey == user.PubKey)
-                .ExecuteUpdateAsync(o => o.SetProperty(v => v.Balance, v => v.Balance - cost));
-        }
-
-        _logger.LogInformation("Stream consumed {n} seconds for {pubkey} costing {cost:#,##0} milli-sats", duration, user.PubKey, cost);
-        if (user.Balance >= balanceAlertThreshold && user.Balance - cost < balanceAlertThreshold)
-        {
-            _nostr.Send(new NostrEventRequest(CreateStreamChat(user,
-                $"Your balance is below {(int)(balanceAlertThreshold / 1000m)} sats, please topup")));
-        }
-
-        if (user.Balance <= 0)
-        {
-            _logger.LogInformation("Kicking stream due to low balance");
-            await _srsApi.KickClient(clientId);
-        }
-    }
-
-    public async Task PatchEvent(string pubkey, string? title, string? summary, string? image, string[]? tags, string? contentWarning)
-    {
-        var user = await _db.Users.SingleOrDefaultAsync(a => a.PubKey == pubkey);
-        if (user == default) throw new Exception("User not found");
-
-        user.Title = title;
-        user.Summary = summary;
-        user.Image = image;
-        user.Tags = tags != null ? string.Join(",", tags) : null;
-        user.ContentWarning = contentWarning;
-
-        var ev = CreateStreamEvent(user);
-        user.Event = JsonConvert.SerializeObject(ev, NostrSerializer.Settings);
-
-        await _db.SaveChangesAsync();
-        _nostr.Send(new NostrEventRequest(ev));
-    }
-
-    public async Task UpdateViewers(string streamKey, int viewers)
-    {
-        var user = await GetUserFromStreamKey(streamKey);
-        if (user == default) throw new Exception("No stream key found");
-
-        var existingEvent = user.GetNostrEvent();
-        if (existingEvent?.Tags != default && (existingEvent.Tags.FindFirstTagValue("status")?.Equals("ended") ?? false)) return;
-
-        var oldViewers = existingEvent?.Tags?.FindFirstTagValue("current_participants");
-        if (string.IsNullOrEmpty(oldViewers) || int.Parse(oldViewers) != viewers)
-        {
-            var ev = CreateStreamEvent(user, viewers: viewers);
-            await PublishEvent(user, ev);
-        }
-    }
-
-    private async Task PublishEvent(User user, NostrEvent ev)
-    {
-        await _db.Users
-            .Where(a => a.PubKey == user.PubKey)
-            .ExecuteUpdateAsync(o => o.SetProperty(v => v.Event, JsonConvert.SerializeObject(ev, NostrSerializer.Settings)));
-
-        _nostr.Send(new NostrEventRequest(ev));
-    }
-
-    private NostrEvent CreateStreamChat(User user, string message)
-    {
-        var pk = NostrPrivateKey.FromBech32(_config.PrivateKey);
-        var ev = new NostrEvent
-        {
-            Kind = StreamChatKind,
-            Content = message,
-            CreatedAt = DateTime.Now,
-            Tags = new NostrEventTags(
-                new NostrEventTag("a", $"{StreamEventKind}:{pk.DerivePublicKey().Hex}:{user.PubKey}")
-            )
-        };
-
-        return ev.Sign(pk);
-    }
-
-    private NostrEvent CreateStreamEvent(User user, string? state = null, int? viewers = null)
-    {
-        var existingEvent = user.GetNostrEvent();
-        var status = state ?? existingEvent?.Tags?.FindFirstTagValue("status") ?? "ended";
-
-        var tags = new List<NostrEventTag>
-        {
-            new("d", user.PubKey),
-            new("title", user.Title ?? ""),
-            new("summary", user.Summary ?? ""),
-            new("streaming", GetStreamUrl(user)),
-            new("image", user.Image ?? ""),
-            new("status", status),
-            new("p", user.PubKey, "", "host"),
-            new("relays", _config.Relays),
-        };
-
-        if (status == "live")
-        {
-            var starts = existingEvent?.Tags?.FindFirstTagValue("starts") ?? DateTimeOffset.UtcNow.ToUnixTimeSeconds().ToString();
-            tags.Add(new("starts", starts));
-            tags.Add(
-                new("current_participants",
-                    (viewers.HasValue ? viewers.ToString() : null) ??
-                    existingEvent?.Tags?.FindFirstTagValue("current_participants") ?? "0"));
-
-            if (!string.IsNullOrEmpty(user.ContentWarning))
-            {
-                tags.Add(new("content-warning", user.ContentWarning));
-            }
-        }
-
-        foreach (var tag in !string.IsNullOrEmpty(user.Tags) ?
-                     user.Tags.Split(",", StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries) : Array.Empty<string>())
-        {
-            tags.Add(new("t", tag));
-        }
-
-        var ev = new NostrEvent
-        {
-            Kind = StreamEventKind,
-            Content = "",
-            CreatedAt = DateTime.Now,
-            Tags = new NostrEventTags(tags)
-        };
-
-        return ev.Sign(NostrPrivateKey.FromBech32(_config.PrivateKey));
-    }
-
-    private string GetStreamUrl(User u)
-    {
-        var ub = new Uri(_config.DataHost, $"{u.PubKey}.m3u8");
-        return ub.ToString();
-    }
-
-    private async Task<User?> GetUserFromStreamKey(string streamKey)
-    {
-        return await _db.Users.SingleOrDefaultAsync(a => a.StreamKey == streamKey);
-    }
-}

NostrStreamer/Services/StreamManager/IStreamManager.cs

@@ -0,0 +1,69 @@
+using NostrStreamer.Database;
+
+namespace NostrStreamer.Services.StreamManager;
+
+public interface IStreamManager
+{
+    /// <summary>
+    /// Return the current stream
+    /// </summary>
+    /// <returns></returns>
+    UserStream GetStream();
+    
+    /// <summary>
+    /// Stream ingress check on srs-edge
+    /// </summary>
+    /// <returns>List of forward URLs</returns>
+    Task<List<string>> OnForward();
+    
+    /// <summary>
+    /// Stream started at origin for HLS split
+    /// </summary>
+    /// <returns></returns>
+    Task StreamStarted();
+    
+    /// <summary>
+    /// Stream stopped
+    /// </summary>
+    /// <returns></returns>
+    Task StreamStopped();
+    
+    /// <summary>
+    /// Stream reap HLS
+    /// </summary>
+    /// <param name="duration"></param>
+    /// <returns></returns>
+    Task ConsumeQuota(double duration);
+
+    /// <summary>
+    /// Update stream details
+    /// </summary>
+    /// <param name="title"></param>
+    /// <param name="summary"></param>
+    /// <param name="image"></param>
+    /// <param name="tags"></param>
+    /// <param name="contentWarning"></param>
+    /// <returns></returns>
+    Task PatchEvent(string? title, string? summary, string? image, string[]? tags, string? contentWarning);
+
+    /// <summary>
+    /// Update viewer count
+    /// </summary>
+    public Task UpdateViewers();
+    
+    /// <summary>
+    /// Add a guest to the stream
+    /// </summary>
+    /// <param name="pubkey"></param>
+    /// <param name="role"></param>
+    /// <param name="zapSplit"></param>
+    /// <returns></returns>
+    Task AddGuest(string pubkey, string role, decimal zapSplit);
+
+    /// <summary>
+    /// Remove guest from the stream
+    /// </summary>
+    /// <param name="pubkey"></param>
+    /// <returns></returns>
+    Task RemoveGuest(string pubkey);
+}

NostrStreamer/Services/StreamManager/LowBalanceException.cs

@@ -0,0 +1,8 @@
+namespace NostrStreamer.Services.StreamManager;
+
+public class LowBalanceException : Exception
+{
+    public LowBalanceException(string message) : base(message)
+    {
+    }
+}

NostrStreamer/Services/StreamManager/NostrStreamManager.cs

@@ -0,0 +1,167 @@
+using Microsoft.EntityFrameworkCore;
+using Newtonsoft.Json;
+using Nostr.Client.Json;
+using NostrStreamer.Database;
+
+namespace NostrStreamer.Services.StreamManager;
+
+public class NostrStreamManager : IStreamManager
+{
+    private readonly ILogger<NostrStreamManager> _logger;
+    private readonly StreamManagerContext _context;
+    private readonly StreamEventBuilder _eventBuilder;
+    private readonly SrsApi _srsApi;
+
+    public NostrStreamManager(ILogger<NostrStreamManager> logger, StreamManagerContext context,
+        StreamEventBuilder eventBuilder, SrsApi srsApi)
+    {
+        _logger = logger;
+        _context = context;
+        _eventBuilder = eventBuilder;
+        _srsApi = srsApi;
+    }
+
+    public UserStream GetStream()
+    {
+        return _context.UserStream;
+    }
+    
+    public Task<List<string>> OnForward()
+    {
+        if (_context.User.Balance <= 0)
+        {
+            throw new LowBalanceException("User balance empty");
+        }
+
+        return Task.FromResult(new List<string>()
+        {
+            $"rtmp://localhost/{_context.UserStream.Endpoint.App}/{_context.User.StreamKey}?vhost={_context.UserStream.Endpoint.Forward}"
+        });
+    }
+
+    public async Task StreamStarted()
+    {
+        _logger.LogInformation("Stream started for: {pubkey}", _context.User.PubKey);
+
+        if (_context.User.Balance <= 0)
+        {
+            throw new Exception("User balance empty");
+        }
+
+        await UpdateStreamState(UserStreamState.Live);
+    }
+
+    public async Task StreamStopped()
+    {
+        _logger.LogInformation("Stream stopped for: {pubkey}", _context.User.PubKey);
+
+        await UpdateStreamState(UserStreamState.Ended);
+    }
+
+    public async Task ConsumeQuota(double duration)
+    {
+        const long balanceAlertThreshold = 500_000;
+        var cost = (long)Math.Ceiling(_context.UserStream.Endpoint.Cost * (duration / 60d));
+        if (cost > 0)
+        {
+            await _context.Db.Users
+                .Where(a => a.PubKey == _context.User.PubKey)
+                .ExecuteUpdateAsync(o => o.SetProperty(v => v.Balance, v => v.Balance - cost));
+        }
+
+        _logger.LogInformation("Stream consumed {n} seconds for {pubkey} costing {cost:#,##0} milli-sats", duration, _context.User.PubKey,
+            cost);
+
+        if (_context.User.Balance >= balanceAlertThreshold && _context.User.Balance - cost < balanceAlertThreshold)
+        {
+            var chat = _eventBuilder.CreateStreamChat(_context.UserStream,
+                $"Your balance is below {(int)(balanceAlertThreshold / 1000m)} sats, please topup");
+
+            _eventBuilder.BroadcastEvent(chat);
+        }
+
+        if (_context.User.Balance <= 0)
+        {
+            _logger.LogInformation("Kicking stream due to low balance");
+            await _srsApi.KickClient(_context.UserStream.ClientId);
+        }
+    }
+
+    public async Task PatchEvent(string? title, string? summary, string? image, string[]? tags, string? contentWarning)
+    {
+        var user = _context.User;
+
+        await _context.Db.Users
+            .Where(a => a.PubKey == _context.User.PubKey)
+            .ExecuteUpdateAsync(o => o.SetProperty(v => v.Title, title)
+                .SetProperty(v => v.Summary, summary)
+                .SetProperty(v => v.Image, image)
+                .SetProperty(v => v.Tags, tags != null ? string.Join(",", tags) : null)
+                .SetProperty(v => v.ContentWarning, contentWarning));
+
+        user.Title = title;
+        user.Summary = summary;
+        user.Image = image;
+        user.Tags = tags != null ? string.Join(",", tags) : null;
+        user.ContentWarning = contentWarning;
+
+        var ev = _eventBuilder.CreateStreamEvent(user, _context.UserStream);
+        await _context.Db.Streams.Where(a => a.Id == _context.UserStream.Id)
+            .ExecuteUpdateAsync(o => o.SetProperty(v => v.Event, JsonConvert.SerializeObject(ev, NostrSerializer.Settings)));
+
+        _eventBuilder.BroadcastEvent(ev);
+    }
+
+    public async Task AddGuest(string pubkey, string role, decimal zapSplit)
+    {
+        _context.Db.Guests.Add(new()
+        {
+            StreamId = _context.UserStream.Id,
+            PubKey = pubkey,
+            Role = role,
+            ZapSplit = zapSplit
+        });
+
+        await _context.Db.SaveChangesAsync();
+    }
+
+    public async Task RemoveGuest(string pubkey)
+    {
+        await _context.Db.Guests
+            .Where(a => a.PubKey == pubkey && a.StreamId == _context.UserStream.Id)
+            .ExecuteDeleteAsync();
+    }
+
+    public async Task UpdateViewers()
+    {
+        if (_context.UserStream.State is not UserStreamState.Live) return;
+
+        var existingEvent = _context.UserStream.GetEvent();
+        var oldViewers = existingEvent?.Tags?.FindFirstTagValue("current_participants");
+
+        var newEvent = _eventBuilder.CreateStreamEvent(_context.User, _context.UserStream);
+        var newViewers = newEvent?.Tags?.FindFirstTagValue("current_participants");
+
+        if (newEvent != default && int.TryParse(oldViewers, out var a) && int.TryParse(newViewers, out var b) && a != b)
+        {
+            await _context.Db.Streams.Where(a => a.Id == _context.UserStream.Id)
+                .ExecuteUpdateAsync(o => o.SetProperty(v => v.Event, JsonConvert.SerializeObject(newEvent, NostrSerializer.Settings)));
+
+            _eventBuilder.BroadcastEvent(newEvent);
+        }
+    }
+
+    private async Task UpdateStreamState(UserStreamState state)
+    {
+        _context.UserStream.State = state;
+        var ev = _eventBuilder.CreateStreamEvent(_context.User, _context.UserStream);
+
+        DateTime? ends = state == UserStreamState.Ended ? DateTime.UtcNow : null;
+        await _context.Db.Streams.Where(a => a.Id == _context.UserStream.Id)
+            .ExecuteUpdateAsync(o => o.SetProperty(v => v.State, state)
+                .SetProperty(v => v.Event, JsonConvert.SerializeObject(ev, NostrSerializer.Settings))
+                .SetProperty(v => v.Ends, ends));
+
+        _eventBuilder.BroadcastEvent(ev);
+    }
+}

NostrStreamer/Services/StreamManager/StreamInfo.cs

@@ -0,0 +1,12 @@
+namespace NostrStreamer.Services.StreamManager;
+
+public class StreamInfo
+{
+    public string App { get; init; } = null!;
+
+    public string Variant { get; init; } = null!;
+
+    public string StreamKey { get; init; } = null!;
+
+    public string ClientId { get; init; } = null!;
+}

NostrStreamer/Services/StreamManager/StreamManagerContext.cs

@@ -0,0 +1,10 @@
+using NostrStreamer.Database;
+
+namespace NostrStreamer.Services.StreamManager;
+
+public class StreamManagerContext
+{
+    public StreamerContext Db { get; init; } = null!;
+    public UserStream UserStream { get; init; } = null!;
+    public User User => UserStream.User;
+}

NostrStreamer/Services/StreamManager/StreamManagerFactory.cs

@@ -0,0 +1,119 @@
+using Microsoft.EntityFrameworkCore;
+using Newtonsoft.Json;
+using Nostr.Client.Json;
+using NostrStreamer.Database;
+
+namespace NostrStreamer.Services.StreamManager;
+
+public class StreamManagerFactory
+{
+    private readonly StreamerContext _db;
+    private readonly ILoggerFactory _loggerFactory;
+    private readonly StreamEventBuilder _eventBuilder;
+    private readonly SrsApi _srsApi;
+
+    public StreamManagerFactory(StreamerContext db, ILoggerFactory loggerFactory, StreamEventBuilder eventBuilder,
+        SrsApi srsApi)
+    {
+        _db = db;
+        _loggerFactory = loggerFactory;
+        _eventBuilder = eventBuilder;
+        _srsApi = srsApi;
+    }
+
+    public async Task<IStreamManager> ForStream(Guid id)
+    {
+        var currentStream = await _db.Streams
+            .AsNoTracking()
+            .Include(a => a.User)
+            .Include(a => a.Endpoint)
+            .FirstOrDefaultAsync(a => a.Id == id);
+
+        if (currentStream == default) throw new Exception("No live stream");
+
+        var ctx = new StreamManagerContext
+        {
+            Db = _db,
+            UserStream = currentStream
+        };
+
+        return new NostrStreamManager(_loggerFactory.CreateLogger<NostrStreamManager>(), ctx, _eventBuilder, _srsApi);
+    }
+    
+    public async Task<IStreamManager> ForCurrentStream(string pubkey)
+    {
+        var currentStream = await _db.Streams
+            .AsNoTracking()
+            .Include(a => a.User)
+            .Include(a => a.Endpoint)
+            .FirstOrDefaultAsync(a => a.PubKey.Equals(pubkey) && a.State == UserStreamState.Live);
+
+        if (currentStream == default) throw new Exception("No live stream");
+
+        var ctx = new StreamManagerContext
+        {
+            Db = _db,
+            UserStream = currentStream
+        };
+
+        return new NostrStreamManager(_loggerFactory.CreateLogger<NostrStreamManager>(), ctx, _eventBuilder, _srsApi);
+    }
+
+    public async Task<IStreamManager> ForStream(StreamInfo info)
+    {
+        var stream = await _db.Streams
+            .AsNoTracking()
+            .Include(a => a.User)
+            .Include(a => a.Endpoint)
+            .FirstOrDefaultAsync(a =>
+                a.ClientId.Equals(info.ClientId) &&
+                a.User.StreamKey.Equals(info.StreamKey) &&
+                a.Endpoint.App.Equals(info.App));
+
+        if (stream == default)
+        {
+            var user = await _db.Users
+                .AsNoTracking()
+                .SingleOrDefaultAsync(a => a.StreamKey.Equals(info.StreamKey));
+
+            if (user == default) throw new Exception("No user found");
+
+            var ep = await _db.Endpoints
+                .AsNoTracking()
+                .SingleOrDefaultAsync(a => a.App.Equals(info.App));
+
+            if (ep == default) throw new Exception("No endpoint found");
+
+            stream = new()
+            {
+                EndpointId = ep.Id,
+                PubKey = user.PubKey,
+                ClientId = info.ClientId,
+                State = UserStreamState.Planned
+            };
+
+            var ev = _eventBuilder.CreateStreamEvent(user, stream);
+            stream.Event = JsonConvert.SerializeObject(ev, NostrSerializer.Settings);
+            _db.Streams.Add(stream);
+            await _db.SaveChangesAsync();
+            
+            // replace again with new values
+            stream = new()
+            {
+                Id = stream.Id,
+                User = user,
+                Endpoint = ep,
+                ClientId = info.ClientId,
+                State = UserStreamState.Planned,
+            };
+        }
+
+        var ctx = new StreamManagerContext
+        {
+            Db = _db,
+            UserStream = stream
+        };
+
+        return new NostrStreamManager(_loggerFactory.CreateLogger<NostrStreamManager>(), ctx, _eventBuilder, _srsApi);
+    }
+}

NostrStreamer/Services/UserBalanceService.cs

@@ -0,0 +1,47 @@
+using System.Security.Cryptography;
+using System.Text;
+using Microsoft.EntityFrameworkCore;
+using Nostr.Client.Utils;
+using NostrStreamer.Database;
+
+namespace NostrStreamer.Services;
+
+public class UserService
+{
+    private readonly StreamerContext _db;
+    private readonly LndNode _lnd;
+
+    public UserService(StreamerContext db, LndNode lnd)
+    {
+        _db = db;
+        _lnd = lnd;
+    }
+
+    public async Task<string> CreateTopup(string pubkey, ulong amount, string? nostr)
+    {
+        var user = await GetUser(pubkey);
+        if (user == default) throw new Exception("No user found");
+
+        var descHash = string.IsNullOrEmpty(nostr) ? null : SHA256.HashData(Encoding.UTF8.GetBytes(nostr)).ToHex();
+        var invoice = await _lnd.AddInvoice(amount * 1000, TimeSpan.FromMinutes(10), $"Top up for {pubkey}", descHash);
+        _db.Payments.Add(new()
+        {
+            PubKey = pubkey,
+            Amount = amount,
+            Invoice = invoice.PaymentRequest,
+            PaymentHash = invoice.RHash.ToByteArray().ToHex(),
+            Nostr = nostr,
+            Type = string.IsNullOrEmpty(nostr) ? PaymentType.Topup : PaymentType.Zap
+        });
+
+        await _db.SaveChangesAsync();
+
+        return invoice.PaymentRequest;
+    }
+
+    public async Task<User?> GetUser(string pubkey)
+    {
+        return await _db.Users.AsNoTracking()
+            .SingleOrDefaultAsync(a => a.PubKey.Equals(pubkey));
+    }
+}

NostrStreamer/Services/ViewCounter.cs

@@ -4,15 +4,15 @@ namespace NostrStreamer.Services;
 
 public class ViewCounter
 {
-    private readonly ConcurrentDictionary<string, Dictionary<string, DateTime>> _sessions = new();
+    private readonly ConcurrentDictionary<Guid, Dictionary<string, DateTime>> _sessions = new();
 
-    public void Activity(string key, string token)
+    public void Activity(Guid id, string token)
     {
-        if (!_sessions.ContainsKey(key))
+        if (!_sessions.ContainsKey(id))
         {
-            _sessions.TryAdd(key, new());
+            _sessions.TryAdd(id, new());
         }
-        if (_sessions.TryGetValue(key, out var x))
+        if (_sessions.TryGetValue(id, out var x))
         {
             x[token] = DateTime.Now;
         }
@@ -36,9 +36,9 @@ public class ViewCounter
         }
     }
 
-    public int Current(string key)
+    public int Current(Guid id)
     {
-        if (_sessions.TryGetValue(key, out var x))
+        if (_sessions.TryGetValue(id, out var x))
         {
             return x.Count;
         }

NostrStreamer/appsettings.json

@@ -11,30 +11,41 @@
     "Database": "User ID=postgres;Password=postgres;Database=streaming;Pooling=true;Host=127.0.0.1:5431"
   },
   "Config": {
-    "RtmpHost": "rtmp://localhost:1935",
+    "RtmpHost": "rtmp://localhost:9005",
-    "SrsHttpHost": "http://localhost:8082",
+    "SrsHttpHost": "http://localhost:9003",
-    "SrsApiHost": "http://localhost:1985",
+    "SrsApiHost": "http://localhost:9002",
     "DataHost": "http://localhost:5295/api/playlist/",
-    "App": "test",
+    "ApiHost": "http://localhost:5295",
-    "Relays": ["ws://localhost:8081"],
+    "Relays": [
+      "ws://localhost:8081"
+    ],
     "PrivateKey": "nsec1yqtv8s8y9krh6l8pwp09lk2jkulr9e0klu95tlk7dgus9cklr4ssdv3d88",
     "Lnd": {
       "Endpoint": "https://localhost:10002",
       "CertPath": "/Users/kieran/.polar/networks/1/volumes/lnd/bob/tls.cert",
       "MacaroonPath": "/Users/kieran/.polar/networks/1/volumes/lnd/bob/data/chain/bitcoin/regtest/admin.macaroon"
     },
-    "Variants": [
+    "Endpoints": [
       {
-        "Name": "source",
+        "Name": "Premium",
-        "Width": 1920,
+        "App": "test",
-        "Height": 1080,
+        "Capabilities": [
-        "Bandwidth": 8000
+          "variant:source",
+          "variant:240h:500",
+          "hls"
+        ],
+        "Forward": "full.in.zap.stream",
+        "Cost": 10000
       },
       {
-        "Name": "240p",
+        "Name": "Passthrough",
-        "Width": 426,
+        "App": "test2",
-        "Height": 240,
+        "Capabilities": [
-        "Bandwidth": 500
+          "variant:source",
+          "hls"
+        ],
+        "Forward": "base.in.zap.stream",
+        "Cost": 2100
       }
     ]
   }

docker-compose.yaml

@@ -1,13 +1,22 @@
 services:
-  srs:
+  srs-origin:
-    image: ossrs/srs
+    image: ossrs/srs:5
     volumes:
-      - "./docker/srs.conf:/usr/local/srs/conf/docker.conf"
+      - "./docker/srs-origin.conf:/usr/local/srs/conf/srs.conf"
     ports:
-      - "1935:1935"
+      - "9001:1935"
-      - "1985:1985"
+      - "9002:1985"
-      - "8082:8080"
+      - "9003:8080"
-      - "8003:8000"
+      - "9004:8000"
+  srs-edge:
+    image: ossrs/srs:5
+    volumes:
+      - "./docker/srs-edge.conf:/usr/local/srs/conf/srs.conf"
+    ports:
+      - "9005:1935"
+      - "9006:1985"
+      - "9007:8080"
+      - "9008:8000"
   nostr:
     image: scsibug/nostr-rs-relay
     ports:

docker/srs-edge.conf

@@ -2,44 +2,31 @@ listen              1935;
 max_connections     1000;
 daemon              off;
 srs_log_tank        console;
+
 http_api {
     enabled         on;
     listen          1985;
 }
+
 http_server {
     enabled         on;
     listen          8080;
 }
+
 rtc_server {
     enabled on;
     listen 8000;
     candidate *;
 }
-vhost transcode {
+
-    hls {
+vhost hls.zap.stream {
-        enabled         on;
+    cluster {
-        hls_dispose     30;
+        mode        remote;
-        hls_fragment    2;
+        origin      srs-origin;
-        hls_window      10;
-    }
-    rtc {
-        enabled     on;
-        rtmp_to_rtc on;
-        rtc_to_rtmp on;
-    }
-    http_remux {
-        enabled     on;
-        mount       [vhost]/[app]/[stream].ts;
-    }
-    http_hooks {
-        enabled     on;
-        on_publish http://10.100.2.226:5295/api/srs;
-        on_unpublish http://10.100.2.226:5295/api/srs;
-        on_hls http://10.100.2.226:5295/api/srs;
     }
 }
 
-vhost __defaultVhost__ {
+vhost base.in.zap.stream {
     transcode {
         enabled     on;
         ffmpeg      ./objs/ffmpeg/bin/ffmpeg;
@@ -48,9 +35,23 @@ vhost __defaultVhost__ {
             enabled         on;
             vcodec          copy;
             acodec          copy;
-            output          rtmp://127.0.0.1:[port]/[app]/[engine]/[stream]?vhost=transcode;
+            output          rtmp://127.0.0.1:[port]/[app]/[engine]/[stream]?vhost=hls.zap.stream;
         }
-        engine 720p {
+    }
+}
+
+vhost full.in.zap.stream {
+    transcode {
+        enabled     on;
+        ffmpeg      ./objs/ffmpeg/bin/ffmpeg;
+        
+        engine source {
+            enabled         on;
+            vcodec          copy;
+            acodec          copy;
+            output          rtmp://127.0.0.1:[port]/[app]/[engine]/[stream]?vhost=hls.zap.stream;
+        }
+        engine 720h {
             enabled         on;
             vcodec          libx264;
             vbitrate        3000;
@@ -68,9 +69,9 @@ vhost __defaultVhost__ {
             abitrate        160;
             asample_rate    44100;
             achannels       2;
-            output          rtmp://127.0.0.1:[port]/[app]/[engine]/[stream]?vhost=transcode;
+            output          rtmp://127.0.0.1:[port]/[app]/[engine]/[stream]?vhost=hls.zap.stream;
         }
-        engine 480p {
+        engine 480h {
             enabled         off;
             vcodec          libx264;
             vbitrate        1000;
@@ -88,9 +89,9 @@ vhost __defaultVhost__ {
             abitrate        96;
             asample_rate    44100;
             achannels       2;
-            output          rtmp://127.0.0.1:[port]/[app]/[engine]/[stream]?vhost=transcode;
+            output          rtmp://127.0.0.1:[port]/[app]/[engine]/[stream]?vhost=hls.zap.stream;
         }
-        engine 240p {
+        engine 240h {
             enabled         off;
             vcodec          libx264;
             vbitrate        500;
@@ -108,7 +109,15 @@ vhost __defaultVhost__ {
             abitrate        72;
             asample_rate    44100;
             achannels       2;
-            output          rtmp://127.0.0.1:[port]/[app]/[engine]/[stream]?vhost=transcode;
+            output          rtmp://127.0.0.1:[port]/[app]/[engine]/[stream]?vhost=hls.zap.stream;
         }
     }
 }
+
+# forward ingest, api decides route
+vhost __defaultVhost__ {
+    forward {
+        enabled on;
+        backend http://10.100.2.226:5295/api/srs;
+    }
+}

docker/srs-origin.conf

@@ -0,0 +1,55 @@
+listen              1935;
+max_connections     1000;
+daemon              off;
+srs_log_tank        console;
+
+http_api {
+    enabled         on;
+    listen          1985;
+}
+
+http_server {
+    enabled         on;
+    listen          8080;
+}
+
+vhost hls.zap.stream {
+    cluster {
+        mode            local;
+    }
+    
+    hls {
+        enabled         on;
+        hls_dispose     30;
+        hls_fragment    2;
+        hls_window      20;
+    }
+    
+    rtc {
+        enabled     on;
+        rtmp_to_rtc on;
+    }
+    
+    http_hooks {
+        enabled     on;
+        on_publish http://10.100.2.226:5295/api/srs;
+        on_unpublish http://10.100.2.226:5295/api/srs;
+        on_hls http://10.100.2.226:5295/api/srs;
+    }
+    
+    transcode {
+        enabled     on;
+        ffmpeg      ./objs/ffmpeg/bin/ffmpeg;
+        
+        engine {
+            enabled         on;
+            vcodec          png;
+            acodec          an;
+            vparams {
+                vframes 1;
+            }
+            oformat         image2;
+            output          ./objs/nginx/html/[app]/[stream].png;
+        }
+    }
+}