-- Instruments (Trading Pairs)
CREATE TABLE instruments (
    id BIGSERIAL PRIMARY KEY,
    entity_id UUID UNIQUE,
    symbol VARCHAR(20) NOT NULL,
    exchange_id BIGINT REFERENCES exchanges(id)
);

-- Price Rules
CREATE TABLE price_rules (
    id BIGSERIAL PRIMARY KEY,
    entity_id UUID UNIQUE,
    name VARCHAR(255) NOT NULL,
    description TEXT,
    is_enabled BOOLEAN DEFAULT true,
    instrument_id BIGINT REFERENCES instruments(id),
    user_id VARCHAR(255),
    last_triggered_at TIMESTAMPTZ,
    last_triggered_price DECIMAL(18,8)
);

-- Price Conditions
CREATE TABLE price_conditions (
    id BIGSERIAL PRIMARY KEY,
    entity_id UUID UNIQUE,
    condition_type VARCHAR(50) NOT NULL,
    value DECIMAL(18,8),
    additional_values JSONB,
    rule_id BIGINT REFERENCES price_rules(id)
);

-- One-Minute Candles (Hypertable)
CREATE TABLE one_min_candle (
    bucket TIMESTAMPTZ NOT NULL,
    symbol VARCHAR(20),
    open DECIMAL(18,8),
    high DECIMAL(18,8),
    low DECIMAL(18,8),
    close DECIMAL(18,8),
    volume DECIMAL(18,8)
);

SELECT create_hypertable('one_min_candle', 'bucket');

db:
  image: timescale/timescaledb-ha:pg15
  restart: always
  user: postgres
  volumes:
    - db-data:/var/lib/postgresql/data
  environment:
    - POSTGRES_DB=price_signal
    - POSTGRES_USER=postgres
    - POSTGRES_PASSWORD=example
  ports:
    - 5432:5432
  healthcheck:
    test: ["CMD", "pg_isready"]
    interval: 10s
    timeout: 5s
    retries: 5

public class NatsService : IPubSub, IAsyncDisposable
{
    private readonly NatsConnection _connection;
    private readonly INatsJSContext _jetStream;
    
    public NatsService(ILogger<NatsService> logger, 
                      IOptions<NatsSettings> natsSettings)
    {
        _connection = new NatsConnection(
            new NatsOpts(){Url = natsSettings.Value.Url});
        _jetStream = new NatsJSContext(_connection);
        
        // Create notifications stream
        _jetStream.CreateStreamAsync(new StreamConfig(
            name: "notifications",
            subjects: new[] { "notifications.>" }
        )).GetAwaiter().GetResult();
    }
    
    public async Task PublishAsync<T>(string subject, T message)
    {
        var content = JsonSerializer.Serialize(message);
        var ack = await _jetStream.PublishAsync(subject, content);
        ack.EnsureSuccess();
    }
    
    public async void Subscribe<T>(string stream, 
                                   Func<T, Task> handler, 
                                   string? subject = null)
    {
        var consumer = await _jetStream
            .CreateOrUpdateConsumerAsync(stream, 
                new ConsumerConfig($"{stream}_processor_dotnet"));
                
        await foreach (var jsMsg in consumer.ConsumeAsync<string>())
        {
            if (jsMsg.Data == null) continue;
            var message = JsonSerializer.Deserialize<T>(jsMsg.Data);
            if (message != null) await handler(message);
            await jsMsg.AckAsync();
        }
    }
}

public class PriceRule : BaseAuditableEntity
{
    public required string Name { get; set; }
    public required string Description { get; set; }
    public bool IsEnabled { get; set; }
    public DateTime? LastTriggeredAt { get; set; }
    public decimal? LastTriggeredPrice { get; set; }
    public NotificationChannelType NotificationChannel { get; set; }
    
    public Instrument Instrument { get; set; }
    public required long InstrumentId { get; set; }
    
    public ICollection<PriceCondition> Conditions { get; set; }
    public User? User { get; set; }
    public string? UserId { get; init; }
    
    public ICollection<PriceRuleTriggerLog> ActivationLogs { get; set; }
}

public class PriceCondition : BaseAuditableEntity
{
    public required string ConditionType { get; set; }
    public decimal Value { get; set; }
    public JsonDocument AdditionalValues { get; set; }
    
    [JsonIgnore]
    public PriceRule Rule { get; set; }
}

public enum ConditionType
{
    PricePercentage,
    Price,
    PriceAction,
    PriceCrossover,
    TechnicalIndicator,
}

public class TelegramNotificationChannel : INotificationChannel
{
    private readonly IPubSub _pubSub;
    public NotificationChannelType ChannelType => 
        NotificationChannelType.telegram;
    
    public async Task SendAsync(string userId, string message)
    {
        var telegramChats = await context.UserNotificationChannels
            .Where(unc => unc.User.Id == userId && 
                         unc.ChannelType == ChannelType)
            .Select(unc => unc.TelegramChatId)
            .ToListAsync();
        
        foreach (var chatId in telegramChats)
        {
            await _pubSub.PublishAsync("notifications.telegram", 
                new Messageinput(chatId.Value, message));
        }
    }
}

func main() {
    // Connect to NATS
    nc, _ := nats.Connect(os.Getenv("NATS_URL"))
    js, _ := jetstream.New(nc)
    
    // Create notifications stream
    s, _ := js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{
        Name:     "notifications",
        Subjects: []string{"notifications.>"},
    })
    
    // Create Telegram bot
    bot, _ := tgbotapi.NewBotAPI(os.Getenv("TELEGRAM_BOT_TOKEN"))
    
    // Consume notification messages
    c, _ := s.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{
        Durable:       "telegram",
        FilterSubject: "notifications.telegram",
        AckPolicy:     jetstream.AckExplicitPolicy,
    })
    
    c.Consume(func(msg jetstream.Msg) {
        var notification Notification
        json.Unmarshal(msg.Data(), &notification)
        
        telegramMsg := tgbotapi.NewMessage(
            notification.ChatID, 
            notification.Message
        )
        bot.Send(telegramMsg)
        msg.Ack()
    })
}

for update := range updates {
    chatID := update.Message.Chat.ID
    username := update.Message.From.UserName
    userId := update.Message.CommandArguments()
    
    chatIDMessage := ChatIDMessage{
        ChatID:   chatID,
        Username: username,
        UserId:   userId,
    }
    
    js.PublishAsync("notifications.init.telegram", 
                   json.Marshal(chatIDMessage))
    
    bot.Send(tgbotapi.NewMessage(chatID, 
        "Hello, your chat ID has been recorded!"))
}

const createRuleMutation = graphql(`
    mutation CreatePriceRule($newRule: PriceRuleInput!) {
        createPriceRule(input:$newRule) {
            id
            name
            description
        }
    }
`)

const onSubmit = (data: z.infer<typeof CreateRuleFormSchema>) => {
    createRule({
        variables: {
            newRule: {
                id: uuidv4(),
                name: data.name,
                description: data.description,
                instrumentId: data.symbol,
                conditions: data.conditions.map(condition => ({
                    conditionType: condition.conditionType,
                    value: condition.threshold, 
                    additionalValues: JSON.stringify({
                        name: condition.indicator, 
                        period: condition.period, 
                        direction: condition.direction
                    })
                }))
            }
        }
    })
}

services:
  server:
    image: nayth/price-signal-graph:latest
    ports:
      - 8080:8080
    environment:
      - Nats__Url=nats://host.docker.internal:4222
      - ASPNETCORE_ENVIRONMENT=Development
  
  db:
    image: timescale/timescaledb-ha:pg15
    environment:
      - POSTGRES_DB=price_signal
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=example
    ports:
      - 5432:5432