import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;
import jakarta.ws.rs.POST;
import jakarta.ws.rs.Path;
import jakarta.ws.rs.Consumes;
import jakarta.ws.rs.core.MediaType;
import jakarta.ws.rs.core.Response;
import org.eclipse.microprofile.reactive.messaging.Channel;
import org.eclipse.microprofile.reactive.messaging.Emitter;

@Path("/orders")
@ApplicationScoped
public class OrderResource {

    // Channel name ต้องตรงกับ mp.messaging.outgoing.orders-out ใน application.properties
    @Inject
    @Channel("orders-out")
    Emitter<String> orderEmitter;

    @POST
    @Consumes(MediaType.APPLICATION_JSON)
    public Response createOrder(OrderRequest req) {
        // สร้าง JSON payload (หรือ serialize ด้วย Jackson)
        String payload = "{\"orderId\":\"" + req.id + "\",\"amount\":" + req.amount + "}";

        // ส่ง message — non-blocking, fire-and-forget
        orderEmitter.send(payload);

        return Response.accepted()
            .entity("{\"status\":\"queued\",\"orderId\":\"" + req.id + "\"}")
            .build();
    }
}

// application.properties
// mp.messaging.outgoing.orders-out.connector=smallrye-rabbitmq
// mp.messaging.outgoing.orders-out.exchange.name=orders-exchange
// mp.messaging.outgoing.orders-out.routing-key=order.created

import org.eclipse.microprofile.reactive.messaging.Incoming;
import org.eclipse.microprofile.reactive.messaging.Message;
import jakarta.enterprise.context.ApplicationScoped;
import java.util.concurrent.CompletionStage;

@ApplicationScoped
public class OrderConsumer {

    // ─── Pattern 1: plain payload (auto-ack on success, auto-nack on exception) ───
    @Incoming("orders-in")
    public void processOrder(String payload) {
        System.out.println("Got order: " + payload);
        // ถ้า throw exception → message จะถูก nack
        // ถ้าสำเร็จ → auto ack
    }

    // ─── Pattern 2: Message<T> wrapper (manual ack/nack) ───────────────────────
    @Incoming("orders-in-manual")
    public CompletionStage<Void> processOrderManual(Message<String> message) {
        String payload = message.getPayload();
        try {
            processBusinessLogic(payload);
            return message.ack();   // บอก RabbitMQ ว่าสำเร็จ → ลบออกจาก queue
        } catch (Exception e) {
            // nack(true) = requeue | nack(false) = drop / ส่งไป DLQ
            return message.nack(e);
        }
    }

    private void processBusinessLogic(String payload) {
        // parse JSON, save to DB, call API, etc.
    }
}

// application.properties
// mp.messaging.incoming.orders-in.connector=smallrye-rabbitmq
// mp.messaging.incoming.orders-in.queue.name=orders
// mp.messaging.incoming.orders-in.queue.durable=true
// mp.messaging.incoming.orders-in.prefetch=10

import io.quarkus.redis.datasource.RedisDataSource;
import io.quarkus.redis.datasource.string.SetArgs;
import io.quarkus.redis.datasource.string.StringCommands;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;

@ApplicationScoped
public class ProductService {

    private final StringCommands<String, String> stringCmds;

    @Inject
    ProductRepository productRepo; // ตัวอย่าง JPA repository

    public ProductService(RedisDataSource ds) {
        // inject RedisDataSource แล้ว get type-safe commands
        this.stringCmds = ds.string(String.class, String.class);
    }

    public String getProduct(String productId) {
        String cacheKey = "product:" + productId;

        // Step 1: ตรวจ Redis cache ก่อน
        String cached = stringCmds.get(cacheKey);
        if (cached != null) {
            System.out.println("Cache HIT for " + cacheKey);
            return cached; // คืนจาก cache ทันที ~1ms
        }

        // Step 2: Cache MISS → ดึงจาก DB
        System.out.println("Cache MISS — querying DB for " + cacheKey);
        Product product = productRepo.findById(productId);
        if (product == null) return null;

        // Step 3: แปลงเป็น JSON string และ store ใน Redis พร้อม TTL 5 นาที
        String json = toJson(product); // Jackson ObjectMapper.writeValueAsString
        SetArgs args = new SetArgs().ex(300); // EX 300 seconds
        stringCmds.set(cacheKey, json, args);

        return json;
    }

    public void invalidateProduct(String productId) {
        // เมื่อ update ข้อมูล → ลบ cache (cache invalidation)
        stringCmds.del("product:" + productId);
    }

    private String toJson(Product p) {
        // Jackson serialize
        return "{\"id\":\"" + p.id + "\",\"name\":\"" + p.name + "\"}";
    }
}

import io.quarkus.redis.datasource.RedisDataSource;
import io.quarkus.redis.datasource.hash.HashCommands;
import io.quarkus.redis.datasource.keys.KeyCommands;
import jakarta.enterprise.context.ApplicationScoped;
import java.util.Map;
import java.util.UUID;

@ApplicationScoped
public class SessionService {

    private final HashCommands<String, String, String> hashCmds;
    private final KeyCommands<String> keyCmds;

    public SessionService(RedisDataSource ds) {
        this.hashCmds = ds.hash(String.class, String.class, String.class);
        this.keyCmds  = ds.key(String.class);
    }

    /** สร้าง session ใหม่หลัง login สำเร็จ */
    public String createSession(String userId, String role) {
        String sessionId = UUID.randomUUID().toString();
        String key = "session:" + sessionId;

        // HSET session:<id> userId "123" role "admin" createdAt "..."
        // hset รับ Map — ไม่มี varargs version ใน Quarkus Redis API
        hashCmds.hset(key, Map.of(
            "userId",    userId,
            "role",      role,
            "createdAt", String.valueOf(System.currentTimeMillis())
        ));

        // ตั้ง TTL 30 นาที ด้วย KeyCommands.expire()
        keyCmds.expire(key, 1800);

        return sessionId;
    }

    /** ดึงข้อมูล session ทั้งหมด */
    public Map<String, String> getSession(String sessionId) {
        return hashCmds.hgetall("session:" + sessionId);
    }

    /** ดึง field เดียว (เร็วกว่า HGETALL) */
    public String getSessionField(String sessionId, String field) {
        return hashCmds.hget("session:" + sessionId, field);
    }

    /** ลบ session เมื่อ logout */
    public void deleteSession(String sessionId) {
        keyCmds.del("session:" + sessionId);
    }
}

import io.quarkus.redis.datasource.RedisDataSource;
import io.quarkus.redis.datasource.keys.KeyCommands;
import io.quarkus.redis.datasource.string.StringCommands;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.ws.rs.WebApplicationException;
import jakarta.ws.rs.core.Response;

@ApplicationScoped
public class RateLimiter {

    private static final int MAX_REQUESTS   = 100; // สูงสุด 100 requests
    private static final int WINDOW_SECONDS = 60;  // ต่อ 1 นาที

    private final StringCommands<String, String> stringCmds;
    private final KeyCommands<String> keyCmds;

    public RateLimiter(RedisDataSource ds) {
        this.stringCmds = ds.string(String.class, String.class);
        this.keyCmds    = ds.key(String.class);
    }

    /**
     * ตรวจสอบ rate limit
     * @param identifier เช่น IP address หรือ userId
     * @throws WebApplicationException 429 Too Many Requests ถ้าเกินลิมิต
     */
    public void checkRateLimit(String identifier) {
        String key = "rate:" + identifier + ":" + getCurrentMinuteBucket();

        // INCR เป็น atomic — ได้ค่า count ใหม่ทันที ไม่มี race condition
        long count = stringCmds.incr(key);

        if (count == 1) {
            // request แรกใน window นี้ → ตั้ง TTL ด้วย KeyCommands
            keyCmds.expire(key, WINDOW_SECONDS);
        }

        if (count > MAX_REQUESTS) {
            throw new WebApplicationException(
                Response.status(429)
                    .header("X-RateLimit-Limit",     MAX_REQUESTS)
                    .header("X-RateLimit-Remaining", 0)
                    .header("Retry-After",           WINDOW_SECONDS)
                    .entity("{\"error\":\"Rate limit exceeded. Try again in " + WINDOW_SECONDS + "s\"}")
                    .build()
            );
        }
    }

    /** Bucket ต่อนาที — key เปลี่ยนทุก 60 วินาทีอัตโนมัติ */
    private String getCurrentMinuteBucket() {
        return String.valueOf(System.currentTimeMillis() / (WINDOW_SECONDS * 1000L));
    }
}

// การใช้งานใน JAX-RS resource:
// @Inject RateLimiter rateLimiter;
//
// @GET @Path("/api/data")
// public Response getData(@Context HttpHeaders headers) {
//     String ip = headers.getRequestHeader("X-Forwarded-For").get(0);
//     rateLimiter.checkRateLimit(ip);
//     // ... proceed
// }