Shahzad Bhatti Welcome to my ramblings and rants!

IDENTIFICATION DIVISION.
PROGRAM-ID. CUSTOMER-INQUIRY.

DATA DIVISION.
WORKING-STORAGE SECTION.
01  CUSTOMER-REC.
    05  CUST-ID        PIC 9(8).
    05  CUST-NAME      PIC X(30).
    05  CUST-BALANCE   PIC 9(7)V99.

LINKAGE SECTION.
01  DFHCOMMAREA.
    05  COMM-CUST-ID   PIC 9(8).

PROCEDURE DIVISION.
    EXEC CICS
        RECEIVE MAP('CUSTMAP')
        MAPSET('CUSTSET')
        INTO(CUSTOMER-REC)
    END-EXEC.
    
    EXEC CICS
        READ FILE('CUSTFILE')
        INTO(CUSTOMER-REC)
        RIDFLD(COMM-CUST-ID)
    END-EXEC.
    
    EXEC CICS
        SEND MAP('RESULTMAP')
        MAPSET('CUSTSET')
        FROM(CUSTOMER-REC)
    END-EXEC.
    
    EXEC CICS RETURN END-EXEC.

CUSTMAP  DFHMSD TYPE=&SYSPARM,                                    X
               MODE=INOUT,                                        X
               LANG=COBOL,                                        X
               CTRL=FREEKB
         DFHMDI SIZE=(24,80)
CUSTID   DFHMDF POS=(05,20),                                      X
               LENGTH=08,                                         X
               ATTRB=(UNPROT,NUM),                                X
               INITIAL='________'
CUSTNAME DFHMDF POS=(07,20),                                      X
               LENGTH=30,                                         X
               ATTRB=PROT

#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#define PORT 8080
#define BUFFER_SIZE 1024

int main() {
    int server_fd, client_fd;
    struct sockaddr_in server_addr, client_addr;
    socklen_t client_len = sizeof(client_addr);
    char buffer[BUFFER_SIZE];
    
    // Create socket
    server_fd = socket(AF_INET, SOCK_STREAM, 0);
    if (server_fd < 0) {
        perror("socket failed");
        exit(EXIT_FAILURE);
    }
    
    // Set socket options to reuse address
    int opt = 1;
    if (setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR, 
                   &opt, sizeof(opt)) < 0) {
        perror("setsockopt failed");
        exit(EXIT_FAILURE);
    }
    
    // Bind to address
    memset(&server_addr, 0, sizeof(server_addr));
    server_addr.sin_family = AF_INET;
    server_addr.sin_addr.s_addr = INADDR_ANY;
    server_addr.sin_port = htons(PORT);
    
    if (bind(server_fd, (struct sockaddr *)&server_addr, 
             sizeof(server_addr)) < 0) {
        perror("bind failed");
        exit(EXIT_FAILURE);
    }
    
    // Listen for connections
    if (listen(server_fd, 10) < 0) {
        perror("listen failed");
        exit(EXIT_FAILURE);
    }
    
    printf("Server listening on port %d\n", PORT);
    
    while (1) {
        // Accept connection
        client_fd = accept(server_fd, 
                          (struct sockaddr *)&client_addr, 
                          &client_len);
        if (client_fd < 0) {
            perror("accept failed");
            continue;
        }
        
        // Read request
        ssize_t bytes_read = recv(client_fd, buffer, 
                                  BUFFER_SIZE - 1, 0);
        if (bytes_read > 0) {
            buffer[bytes_read] = '\0';
            printf("Received: %s\n", buffer);
            
            // Send response
            const char *response = "Message received\n";
            send(client_fd, response, strlen(response), 0);
        }
        
        close(client_fd);
    }
    
    close(server_fd);
    return 0;
}

/* Define a structure for customer data */
struct customer {
    int customer_id;
    string name<30>;
    float balance;
};

/* Define the RPC program */
program CUSTOMER_PROG {
    version CUSTOMER_VERS {
        int ADD_CUSTOMER(customer) = 1;
        customer GET_CUSTOMER(int) = 2;
    } = 1;
} = 0x20000001;

$ rpcgen types.x

#include "types.h"

int *add_customer_1_svc(customer *cust, struct svc_req *rqstp) {
    static int result;
    
    // Add customer to database
    printf("Adding customer: %s (ID: %d)\n", 
           cust->name, cust->customer_id);
    
    result = 1;  // Success
    return &result;
}

customer *get_customer_1_svc(int *cust_id, struct svc_req *rqstp) {
    static customer result;
    
    // Fetch from database
    result.customer_id = *cust_id;
    result.name = strdup("John Doe");
    result.balance = 1000.50;
    
    return &result;
}

#include "types.h"

int main(int argc, char *argv[]) {
    CLIENT *clnt;
    customer cust;
    int *result;
    
    clnt = clnt_create("localhost", CUSTOMER_PROG, 
                       CUSTOMER_VERS, "tcp");
    if (clnt == NULL) {
        clnt_pcreateerror("localhost");
        exit(1);
    }
    
    // Call remote procedure
    cust.customer_id = 123;
    cust.name = "Alice Smith";
    cust.balance = 5000.00;
    
    result = add_customer_1(&cust, clnt);
    if (result == NULL) {
        clnt_perror(clnt, "call failed");
    }
    
    clnt_destroy(clnt);
    return 0;
}

#include <mpi.h>
#include <stdio.h>
#include <stdlib.h>

#define ARRAY_SIZE 1000

int main(int argc, char** argv) {
    int rank, size;
    int data[ARRAY_SIZE];
    int local_sum = 0, global_sum = 0;
    int chunk_size, start, end;
    
    MPI_Init(&argc, &argv);
    MPI_Comm_rank(MPI_COMM_WORLD, &rank);
    MPI_Comm_size(MPI_COMM_WORLD, &size);
    
    // Initialize data on root
    if (rank == 0) {
        for (int i = 0; i < ARRAY_SIZE; i++) {
            data[i] = i + 1;
        }
    }
    
    // Broadcast data to all processes
    MPI_Bcast(data, ARRAY_SIZE, MPI_INT, 0, MPI_COMM_WORLD);
    
    // Each process computes sum of its chunk
    chunk_size = ARRAY_SIZE / size;
    start = rank * chunk_size;
    end = (rank == size - 1) ? ARRAY_SIZE : start + chunk_size;
    
    for (int i = start; i < end; i++) {
        local_sum += data[i];
    }
    
    // Reduce all local sums to global sum
    MPI_Reduce(&local_sum, &global_sum, 1, MPI_INT, 
               MPI_SUM, 0, MPI_COMM_WORLD);
    
    if (rank == 0) {
        printf("Global sum: %d\n", global_sum);
    }
    
    MPI_Finalize();
    return 0;
}

public class SumTask extends ActiveEntity {
    public Object execute(Object arg, JavaNOWAPI api) {
        Integer myId = (Integer) arg;
        EntitySpace workspace = new EntitySpace("RESULTS");
        
        // Compute partial sum
        int partialSum = 0;
        for (int i = myId * 100; i < (myId + 1) * 100; i++) {
            partialSum += i;
        }
        
        // Store result in EntitySpace
        return new Integer(partialSum);
    }
}

// Main application
public class ParallelSum extends JavaNOWApplication {
    public void master() {
        EntitySpace workspace = new EntitySpace("RESULTS");
        
        // Spawn parallel tasks
        for (int i = 0; i < 10; i++) {
            ActiveEntity task = new SumTask(new Integer(i));
            getJavaNOWAPI().eval(workspace, task, new Integer(i));
        }
        
        // Collect results
        int totalSum = 0;
        for (int i = 0; i < 10; i++) {
            Entity result = getJavaNOWAPI().get(
                workspace, new Entity(new Integer(i)));
            totalSum += ((Integer)result.getEntityValue()).intValue();
        }
        
        System.out.println("Total sum: " + totalSum);
    }
    
    public void slave(int id) {
        // Slave nodes wait for work
    }
}

#include <nwcalls.h>
#include <nwipxspx.h>

int main() {
    IPXAddress server_addr;
    IPXPacket packet;
    WORD socket_number = 0x4000;
    
    // Open IPX socket
    IPXOpenSocket(socket_number, 0);
    
    // Setup address
    memset(&server_addr, 0, sizeof(IPXAddress));
    memcpy(server_addr.network, target_network, 4);
    memcpy(server_addr.node, target_node, 6);
    server_addr.socket = htons(socket_number);
    
    // Send packet
    packet.packetType = 4;  // IPX packet type
    memcpy(packet.data, "Hello", 5);
    IPXSendPacket(socket_number, &server_addr, &packet);
    
    IPXCloseSocket(socket_number);
    return 0;
}

#!/usr/bin/perl
use strict;
use warnings;
use CGI;

my $cgi = CGI->new;

print $cgi->header('text/html');
print "<html><body>\n";
print "<h1>Hello from CGI!</h1>\n";

my $name = $cgi->param('name') || 'Guest';
print "<p>Welcome, $name!</p>\n";

# Simulate database query
my $user_count = 42;
print "<p>Total users: $user_count</p>\n";

print "</body></html>\n";

#include <stdio.h>
#include <stdlib.h>
#include <string.h>

int main() {
    char *query_string = getenv("QUERY_STRING");
    
    printf("Content-Type: text/html\n\n");
    printf("<html><body>\n");
    printf("<h1>CGI in C</h1>\n");
    
    if (query_string) {
        printf("<p>Query string: %s</p>\n", query_string);
    }
    
    printf("</body></html>\n");
    return 0;
}

import java.io.*;
import java.net.*;

public class SimpleServer {
    public static void main(String[] args) throws IOException {
        int port = 8080;
        
        try (ServerSocket serverSocket = new ServerSocket(port)) {
            System.out.println("Server listening on port " + port);
            
            while (true) {
                try (Socket clientSocket = serverSocket.accept();
                     BufferedReader in = new BufferedReader(
                         new InputStreamReader(clientSocket.getInputStream()));
                     PrintWriter out = new PrintWriter(
                         clientSocket.getOutputStream(), true)) {
                    
                    String request = in.readLine();
                    System.out.println("Received: " + request);
                    
                    out.println("Message received");
                }
            }
        }
    }
}

public class ConcurrentServer {
    public static void main(String[] args) throws IOException {
        ServerSocket serverSocket = new ServerSocket(8080);
        
        while (true) {
            Socket clientSocket = serverSocket.accept();
            
            // Spawn thread to handle client
            new Thread(new ClientHandler(clientSocket)).start();
        }
    }
    
    static class ClientHandler implements Runnable {
        private Socket socket;
        
        public ClientHandler(Socket socket) {
            this.socket = socket;
        }
        
        public void run() {
            try (BufferedReader in = new BufferedReader(
                     new InputStreamReader(socket.getInputStream()));
                 PrintWriter out = new PrintWriter(
                     socket.getOutputStream(), true)) {
                
                String request = in.readLine();
                // Process request
                out.println("Response");
                
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                try { socket.close(); } catch (IOException e) {}
            }
        }
    }
}

public class ThreadSafeCounter {
    private int count = 0;
    
    public synchronized void increment() {
        count++;
    }
    
    public synchronized int getCount() {
        return count;
    }
}

import java.rmi.Remote;
import java.rmi.RemoteException;

public interface Calculator extends Remote {
    int add(int a, int b) throws RemoteException;
    int multiply(int a, int b) throws RemoteException;
}

import java.rmi.server.UnicastRemoteObject;
import java.rmi.RemoteException;

public class CalculatorImpl extends UnicastRemoteObject 
                            implements Calculator {
    
    public CalculatorImpl() throws RemoteException {
        super();
    }
    
    public int add(int a, int b) throws RemoteException {
        return a + b;
    }
    
    public int multiply(int a, int b) throws RemoteException {
        return a * b;
    }
}

import java.rmi.Naming;
import java.rmi.registry.LocateRegistry;

public class Server {
    public static void main(String[] args) {
        try {
            LocateRegistry.createRegistry(1099);
            Calculator calc = new CalculatorImpl();
            Naming.rebind("Calculator", calc);
            System.out.println("Server ready");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

import java.rmi.Naming;

public class Client {
    public static void main(String[] args) {
        try {
            Calculator calc = (Calculator) Naming.lookup(
                "rmi://localhost/Calculator");
            
            int result = calc.add(5, 3);
            System.out.println("5 + 3 = " + result);
            
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

import net.jini.core.lookup.ServiceRegistrar;
import net.jini.discovery.LookupDiscovery;
import net.jini.lease.LeaseRenewalManager;
import java.rmi.Remote;
import java.rmi.RemoteException;

// Service interface
public interface BookingService extends Remote {
    String searchFlights(String origin, String destination) 
        throws RemoteException;
    boolean bookFlight(String flightId, String passenger) 
        throws RemoteException;
}

// Service provider
public class BookingServiceProvider implements DiscoveryListener {
    
    public void discovered(DiscoveryEvent event) {
        ServiceRegistrar[] registrars = event.getRegistrars();
        
        for (ServiceRegistrar registrar : registrars) {
            try {
                BookingService service = new BookingServiceImpl();
                Entry[] attributes = new Entry[] {
                    new Name("FlightBookingService")
                };
                
                ServiceItem item = new ServiceItem(null, service, attributes);
                ServiceRegistration reg = registrar.register(
                    item, Lease.FOREVER);
                
                // Auto-renew lease
                leaseManager.renewUntil(reg.getLease(), Lease.FOREVER, null);
                
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

public class BookingClient implements DiscoveryListener {
    
    public void discovered(DiscoveryEvent event) {
        ServiceRegistrar[] registrars = event.getRegistrars();
        
        for (ServiceRegistrar registrar : registrars) {
            try {
                ServiceTemplate template = new ServiceTemplate(
                    null, new Class[] { BookingService.class }, null);
                
                ServiceItem item = registrar.lookup(template);
                
                if (item != null) {
                    BookingService booking = (BookingService) item.service;
                    String flights = booking.searchFlights("SFO", "NYC");
                    booking.bookFlight("FL123", "John Smith");
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

module TrafficMonitor {
    struct SensorData {
        long sensor_id;
        float speed;
        long timestamp;
    };
    
    typedef sequence<SensorData> SensorDataList;
    
    interface TrafficService {
        void reportData(in SensorData data);
        SensorDataList getRecentData(in long minutes);
        float getAverageSpeed();
    };
};

$ idl traffic.idl

#include "TrafficService_impl.h"
#include <iostream>
#include <vector>

class TrafficServiceImpl : public POA_TrafficMonitor::TrafficService {
private:
    std::vector<TrafficMonitor::SensorData> data_store;
    
public:
    void reportData(const TrafficMonitor::SensorData& data) {
        data_store.push_back(data);
        std::cout << "Received data from sensor " 
                  << data.sensor_id << std::endl;
    }
    
    TrafficMonitor::SensorDataList* getRecentData(CORBA::Long minutes) {
        TrafficMonitor::SensorDataList* result = 
            new TrafficMonitor::SensorDataList();
        
        // Filter data from last N minutes
        time_t cutoff = time(NULL) - (minutes * 60);
        for (const auto& entry : data_store) {
            if (entry.timestamp >= cutoff) {
                result->length(result->length() + 1);
                (*result)[result->length() - 1] = entry;
            }
        }
        return result;
    }
    
    CORBA::Float getAverageSpeed() {
        if (data_store.empty()) return 0.0;
        
        float sum = 0.0;
        for (const auto& entry : data_store) {
            sum += entry.speed;
        }
        return sum / data_store.size();
    }
};

import org.omg.CORBA.*;
import TrafficMonitor.*;

public class TrafficClient {
    public static void main(String[] args) {
        try {
            // Initialize ORB
            ORB orb = ORB.init(args, null);
            
            // Get reference to service
            org.omg.CORBA.Object obj = 
                orb.string_to_object("corbaname::localhost:1050#TrafficService");
            TrafficService service = TrafficServiceHelper.narrow(obj);
            
            // Report sensor data
            SensorData data = new SensorData();
            data.sensor_id = 101;
            data.speed = 65.5f;
            data.timestamp = (int)(System.currentTimeMillis() / 1000);
            
            service.reportData(data);
            
            // Get average speed
            float avgSpeed = service.getAverageSpeed();
            System.out.println("Average speed: " + avgSpeed + " mph");
            
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

<?xml version="1.0"?>
<definitions name="CustomerService"
   targetNamespace="http://example.com/customer"
   xmlns="http://schemas.xmlsoap.org/wsdl/"
   xmlns:soap="http://schemas.xmlsoap.org/wsdl/soap/"
   xmlns:tns="http://example.com/customer"
   xmlns:xsd="http://www.w3.org/2001/XMLSchema">
   
   <types>
      <xsd:schema targetNamespace="http://example.com/customer">
         <xsd:complexType name="Customer">
            <xsd:sequence>
               <xsd:element name="id" type="xsd:int"/>
               <xsd:element name="name" type="xsd:string"/>
               <xsd:element name="balance" type="xsd:double"/>
            </xsd:sequence>
         </xsd:complexType>
      </xsd:schema>
   </types>
   
   <message name="GetCustomerRequest">
      <part name="customerId" type="xsd:int"/>
   </message>
   
   <message name="GetCustomerResponse">
      <part name="customer" type="tns:Customer"/>
   </message>
   
   <portType name="CustomerPortType">
      <operation name="getCustomer">
         <input message="tns:GetCustomerRequest"/>
         <output message="tns:GetCustomerResponse"/>
      </operation>
   </portType>
   
   <binding name="CustomerBinding" type="tns:CustomerPortType">
      <soap:binding transport="http://schemas.xmlsoap.org/soap/http"/>
      <operation name="getCustomer">
         <soap:operation soapAction="getCustomer"/>
         <input>
            <soap:body use="literal"/>
         </input>
         <output>
            <soap:body use="literal"/>
         </output>
      </operation>
   </binding>
   
   <service name="CustomerService">
      <port name="CustomerPort" binding="tns:CustomerBinding">
         <soap:address location="http://example.com/customer"/>
      </port>
   </service>
</definitions>

<?xml version="1.0"?>
<soap:Envelope 
    xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/"
    xmlns:cust="http://example.com/customer">
  <soap:Header>
    <cust:Authentication>
      <cust:username>john</cust:username>
      <cust:password>secret</cust:password>
    </cust:Authentication>
  </soap:Header>
  <soap:Body>
    <cust:getCustomer>
      <cust:customerId>12345</cust:customerId>
    </cust:getCustomer>
  </soap:Body>
</soap:Envelope>

<?xml version="1.0"?>
<soap:Envelope 
    xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/"
    xmlns:cust="http://example.com/customer">
  <soap:Body>
    <cust:getCustomerResponse>
      <cust:customer>
        <cust:id>12345</cust:id>
        <cust:name>John Smith</cust:name>
        <cust:balance>5000.00</cust:balance>
      </cust:customer>
    </cust:getCustomerResponse>
  </soap:Body>
</soap:Envelope>

import javax.servlet.*;
import javax.servlet.http.*;
import java.io.*;

public class CustomerServlet extends HttpServlet {
    
    protected void doGet(HttpServletRequest request, 
                        HttpServletResponse response)
            throws ServletException, IOException {
        
        String customerId = request.getParameter("id");
        
        response.setContentType("application/json");
        PrintWriter out = response.getWriter();
        
        // Fetch customer data
        Customer customer = getCustomerFromDatabase(customerId);
        
        if (customer != null) {
            out.println(String.format(
                "{\"id\": \"%s\", \"name\": \"%s\", \"balance\": %.2f}",
                customer.getId(), customer.getName(), customer.getBalance()
            ));
        } else {
            response.setStatus(HttpServletResponse.SC_NOT_FOUND);
            out.println("{\"error\": \"Customer not found\"}");
        }
    }
    
    protected void doPost(HttpServletRequest request, 
                         HttpServletResponse response)
            throws ServletException, IOException {
        
        BufferedReader reader = request.getReader();
        StringBuilder json = new StringBuilder();
        String line;
        while ((line = reader.readLine()) != null) {
            json.append(line);
        }
        
        // Parse JSON and create customer
        Customer customer = parseJsonToCustomer(json.toString());
        saveCustomerToDatabase(customer);
        
        response.setStatus(HttpServletResponse.SC_CREATED);
        response.setContentType("application/json");
        PrintWriter out = response.getWriter();
        out.println(json.toString());
    }
}

import javax.servlet.*;
import javax.servlet.http.*;
import java.io.IOException;

public class AuthenticationFilter implements Filter {
    
    public void doFilter(ServletRequest request, 
                        ServletResponse response,
                        FilterChain chain) 
            throws IOException, ServletException {
        
        HttpServletRequest httpRequest = (HttpServletRequest) request;
        HttpServletResponse httpResponse = (HttpServletResponse) response;
        
        // Check for authentication token
        String token = httpRequest.getHeader("Authorization");
        
        if (token == null || !isValidToken(token)) {
            httpResponse.setStatus(HttpServletResponse.SC_UNAUTHORIZED);
            httpResponse.getWriter().println("{\"error\": \"Unauthorized\"}");
            return;
        }
        
        // Pass to next filter or servlet
        chain.doFilter(request, response);
    }
    
    private boolean isValidToken(String token) {
        // Validate token
        return token.startsWith("Bearer ") && 
               validateJWT(token.substring(7));
    }
}

<filter>
    <filter-name>AuthenticationFilter</filter-name>
    <filter-class>com.example.AuthenticationFilter</filter-class>
</filter>

<filter-mapping>
    <filter-name>AuthenticationFilter</filter-name>
    <url-pattern>/api/*</url-pattern>
</filter-mapping>

// Remote interface
public interface Customer extends EJBObject {
    String getName() throws RemoteException;
    void setName(String name) throws RemoteException;
    double getBalance() throws RemoteException;
    void setBalance(double balance) throws RemoteException;
}

// Home interface
public interface CustomerHome extends EJBHome {
    Customer create(Integer id, String name) throws CreateException, RemoteException;
    Customer findByPrimaryKey(Integer id) throws FinderException, RemoteException;
}

// Bean implementation
public class CustomerBean implements EntityBean {
    private Integer id;
    private String name;
    private double balance;
    
    public String getName() { return name; }
    public void setName(String name) { this.name = name; }
    public double getBalance() { return balance; }
    public void setBalance(double balance) { this.balance = balance; }
    
    // Container callbacks
    public void ejbActivate() {}
    public void ejbPassivate() {}
    public void ejbLoad() {}
    public void ejbStore() {}
    public void setEntityContext(EntityContext ctx) {}
    public void unsetEntityContext() {}
    
    public Integer ejbCreate(Integer id, String name) {
        this.id = id;
        this.name = name;
        this.balance = 0.0;
        return null;
    }
    
    public void ejbPostCreate(Integer id, String name) {}
}

CustomerHome home = // ... lookup
Customer customer = home.findByPrimaryKey(customerId);

// Each getter is a remote call!
String name = customer.getName();        // Network call
double balance = customer.getBalance();  // Network call

Collection customers = home.findAll();
double totalBalance = 0.0;
for (Customer customer : customers) {
    // Remote call for EVERY iteration!
    totalBalance += customer.getBalance();
}

import requests
import xml.etree.ElementTree as ET

# Create XML request
root = ET.Element('getCustomer')
ET.SubElement(root, 'customerId').text = '12345'
xml_data = ET.tostring(root, encoding='utf-8')

# Send HTTP POST
response = requests.post(
    'http://api.example.com/customer',
    data=xml_data,
    headers={'Content-Type': 'application/xml'}
)

# Parse response
response_tree = ET.fromstring(response.content)
name = response_tree.find('name').text

from flask import Flask, jsonify, request

app = Flask(__name__)

# In-memory data store
customers = {
    '12345': {'id': '12345', 'name': 'John Smith', 'balance': 5000.00}
}

@app.route('/customers/<customer_id>', methods=['GET'])
def get_customer(customer_id):
    customer = customers.get(customer_id)
    if customer:
        return jsonify(customer), 200
    return jsonify({'error': 'Customer not found'}), 404

@app.route('/customers', methods=['POST'])
def create_customer():
    data = request.get_json()
    customer_id = data['id']
    customers[customer_id] = data
    return jsonify(data), 201

@app.route('/customers/<customer_id>', methods=['PUT'])
def update_customer(customer_id):
    if customer_id not in customers:
        return jsonify({'error': 'Customer not found'}), 404
    
    data = request.get_json()
    customers[customer_id].update(data)
    return jsonify(customers[customer_id]), 200

@app.route('/customers/<customer_id>', methods=['DELETE'])
def delete_customer(customer_id):
    if customer_id in customers:
        del customers[customer_id]
        return '', 204
    return jsonify({'error': 'Customer not found'}), 404

if __name__ == '__main__':
    app.run(debug=True)

import requests

# GET customer
response = requests.get('http://localhost:5000/customers/12345')
if response.status_code == 200:
    customer = response.json()
    print(f"Customer: {customer['name']}")

# Create new customer
new_customer = {
    'id': '67890',
    'name': 'Alice Johnson',
    'balance': 3000.00
}
response = requests.post(
    'http://localhost:5000/customers',
    json=new_customer
)

# Update customer
update_data = {'balance': 3500.00}
response = requests.put(
    'http://localhost:5000/customers/67890',
    json=update_data
)

# Delete customer
response = requests.delete('http://localhost:5000/customers/67890')

{
  "id": "12345",
  "name": "John Smith",
  "balance": 5000.00,
  "_links": {
    "self": {"href": "/customers/12345"},
    "orders": {"href": "/customers/12345/orders"},
    "transactions": {"href": "/customers/12345/transactions"}
  }
}

<?xml version="1.0"?>
<customer>
    <id>12345</id>
    <name>John Smith</name>
    <balance>5000.00</balance>
    <orders>
        <order>
            <id>001</id>
            <date>2024-01-15</date>
            <total>99.99</total>
        </order>
        <order>
            <id>002</id>
            <date>2024-02-20</date>
            <total>149.50</total>
        </order>
    </orders>
</customer>

{
  "id": "12345",
  "name": "John Smith",
  "balance": 5000.00,
  "orders": [
    {
      "id": "001",
      "date": "2024-01-15",
      "total": 99.99
    },
    {
      "id": "002",
      "date": "2024-02-20",
      "total": 149.50
    }
  ]
}

{
  "id": "A",
  "children": [
    {
      "id": "B",
      "parent_id": "A"
    }
  ]
}

-module(customer_server).
-export([start/0, init/0, get_customer/1, update_balance/2]).

% Start the server
start() ->
    Pid = spawn(customer_server, init, []),
    register(customer_server, Pid),
    Pid.

% Initialize with empty state
init() ->
    State = #{},  % Empty map
    loop(State).

% Main loop - handle messages
loop(State) ->
    receive
        {get_customer, CustomerId, From} ->
            Customer = maps:get(CustomerId, State, not_found),
            From ! {customer, Customer},
            loop(State);
        
        {update_balance, CustomerId, NewBalance, From} ->
            Customer = maps:get(CustomerId, State),
            UpdatedCustomer = Customer#{balance => NewBalance},
            NewState = maps:put(CustomerId, UpdatedCustomer, State),
            From ! {ok, updated},
            loop(NewState);
        
        {add_customer, CustomerId, Customer, From} ->
            NewState = maps:put(CustomerId, Customer, State),
            From ! {ok, added},
            loop(NewState);
        
        stop ->
            ok;
        
        _ ->
            loop(State)
    end.

% Client functions
get_customer(CustomerId) ->
    customer_server ! {get_customer, CustomerId, self()},
    receive
        {customer, Customer} -> Customer
    after 5000 ->
        timeout
    end.

update_balance(CustomerId, NewBalance) ->
    customer_server ! {update_balance, CustomerId, NewBalance, self()},
    receive
        {ok, updated} -> ok
    after 5000 ->
        timeout
    end.

-module(customer_supervisor).
-behaviour(supervisor).

-export([start_link/0, init/1]).

start_link() ->
    supervisor:start_link({local, ?MODULE}, ?MODULE, []).

init([]) ->
    % Supervisor strategy
    SupFlags = #{
        strategy => one_for_one,  % Restart only failed child
        intensity => 5,            % Max 5 restarts
        period => 60               % Per 60 seconds
    },
    
    % Child specifications
    ChildSpecs = [
        #{
            id => customer_server,
            start => {customer_server, start, []},
            restart => permanent,   % Always restart
            shutdown => 5000,
            type => worker,
            modules => [customer_server]
        },
        #{
            id => order_server,
            start => {order_server, start, []},
            restart => permanent,
            shutdown => 5000,
            type => worker,
            modules => [order_server]
        }
    ],
    
    {ok, {SupFlags, ChildSpecs}}.

% On node1@host1
RemotePid = spawn('node2@host2', module, function, [args]),
RemotePid ! {message, data}.

% On node2@host2 - receives the message
receive
    {message, Data} -> 
        io:format("Received: ~p~n", [Data])
end.

import javax.jms.*;
import javax.naming.*;

public class OrderProducer {
    public static void main(String[] args) throws Exception {
        Context ctx = new InitialContext();
        ConnectionFactory factory = 
            (ConnectionFactory) ctx.lookup("ConnectionFactory");
        Queue queue = (Queue) ctx.lookup("OrderQueue");
        
        Connection connection = factory.createConnection();
        Session session = connection.createSession(
            false, Session.AUTO_ACKNOWLEDGE);
        MessageProducer producer = session.createProducer(queue);
        
        // Create message
        TextMessage message = session.createTextMessage();
        message.setText("{ \"orderId\": \"12345\", " +
                       "\"customerId\": \"67890\", " +
                       "\"amount\": 99.99 }");
        
        // Send message
        producer.send(message);
        System.out.println("Order sent: " + message.getText());
        
        connection.close();
    }
}

import javax.jms.*;
import javax.naming.*;

public class OrderConsumer implements MessageListener {
    
    public static void main(String[] args) throws Exception {
        Context ctx = new InitialContext();
        ConnectionFactory factory = 
            (ConnectionFactory) ctx.lookup("ConnectionFactory");
        Queue queue = (Queue) ctx.lookup("OrderQueue");
        
        Connection connection = factory.createConnection();
        Session session = connection.createSession(
            false, Session.AUTO_ACKNOWLEDGE);
        MessageConsumer consumer = session.createConsumer(queue);
        
        consumer.setMessageListener(new OrderConsumer());
        connection.start();
        
        System.out.println("Waiting for messages...");
        Thread.sleep(Long.MAX_VALUE);  // Keep running
    }
    
    public void onMessage(Message message) {
        try {
            TextMessage textMessage = (TextMessage) message;
            System.out.println("Received order: " + 
                             textMessage.getText());
            
            // Process order
            processOrder(textMessage.getText());
            
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
    
    private void processOrder(String orderJson) {
        // Business logic here
    }
}

@Aspect
@Component
public class LoggingAspect {
    
    private static final Logger logger = 
        LoggerFactory.getLogger(LoggingAspect.class);
    
    @Before("execution(* com.example.service.*.*(..))")
    public void logBefore(JoinPoint joinPoint) {
        logger.info("Executing: " + 
                   joinPoint.getSignature().getName());
    }
    
    @AfterReturning(
        pointcut = "execution(* com.example.service.*.*(..))",
        returning = "result")
    public void logAfterReturning(JoinPoint joinPoint, Object result) {
        logger.info("Method " + 
                   joinPoint.getSignature().getName() + 
                   " returned: " + result);
    }
    
    @Around("@annotation(com.example.Monitored)")
    public Object measureTime(ProceedingJoinPoint joinPoint) 
            throws Throwable {
        long start = System.currentTimeMillis();
        Object result = joinPoint.proceed();
        long time = System.currentTimeMillis() - start;
        logger.info(joinPoint.getSignature().getName() + 
                   " took " + time + " ms");
        return result;
    }
}

// Spring configuration
@Configuration
public class AppConfig {
    
    @Bean
    public CustomerService customerService() {
        return new CustomerServiceImpl(customerRepository());
    }
    
    @Bean
    public CustomerRepository customerRepository() {
        return new DatabaseCustomerRepository(dataSource());
    }
    
    @Bean
    public DataSource dataSource() {
        DriverManagerDataSource ds = new DriverManagerDataSource();
        ds.setDriverClassName("com.mysql.jdbc.Driver");
        ds.setUrl("jdbc:mysql://localhost/mydb");
        return ds;
    }
}

// Service class
@Service
public class CustomerServiceImpl implements CustomerService {
    private final CustomerRepository repository;
    
    @Autowired
    public CustomerServiceImpl(CustomerRepository repository) {
        this.repository = repository;
    }
    
    @Transactional
    public void updateBalance(String customerId, double newBalance) {
        Customer customer = repository.findById(customerId);
        customer.setBalance(newBalance);
        repository.save(customer);
    }
}

// Server configuration
@Configuration
public class ServerConfig {
    
    @Bean
    public HttpInvokerServiceExporter customerService() {
        HttpInvokerServiceExporter exporter = 
            new HttpInvokerServiceExporter();
        exporter.setService(customerServiceImpl());
        exporter.setServiceInterface(CustomerService.class);
        return exporter;
    }
}

// Client configuration
@Configuration
public class ClientConfig {
    
    @Bean
    public HttpInvokerProxyFactoryBean customerService() {
        HttpInvokerProxyFactoryBean proxy = 
            new HttpInvokerProxyFactoryBean();
        proxy.setServiceUrl("http://localhost:8080/customer");
        proxy.setServiceInterface(CustomerService.class);
        return proxy;
    }
}

// Coral IDL
namespace com.amazon.example

structure CustomerData {
    1: required integer customerId
    2: required string name
    3: optional double balance
}

service CustomerService {
    CustomerData getCustomer(1: integer customerId)
    void updateCustomer(1: CustomerData customer)
    list<CustomerData> listCustomers()
}

namespace com.example

service CustomerService {
    version: "2024-01-01"
    operations: [
        GetCustomer
        UpdateCustomer
        ListCustomers
    ]
}

@readonly
operation GetCustomer {
    input: GetCustomerInput
    output: GetCustomerOutput
    errors: [CustomerNotFound]
}

structure GetCustomerInput {
    @required
    customerId: String
}

structure GetCustomerOutput {
    @required
    customer: Customer
}

structure Customer {
    @required
    customerId: String
    
    @required
    name: String
    
    balance: Double
}

@error("client")
structure CustomerNotFound {
    @required
    message: String
}

// Client-side long polling
function pollServer() {
    fetch('/api/events')
        .then(response => response.json())
        .then(data => {
            console.log('Received event:', data);
            updateUI(data);
            
            // Immediately poll again
            pollServer();
        })
        .catch(error => {
            console.error('Polling error:', error);
            // Retry after delay
            setTimeout(pollServer, 5000);
        });
}

pollServer();

const express = require('express');
const app = express();

let pendingRequests = [];

app.get('/api/events', (req, res) => {
    // Hold request open
    pendingRequests.push(res);
    
    // Timeout after 30 seconds
    setTimeout(() => {
        const index = pendingRequests.indexOf(res);
        if (index !== -1) {
            pendingRequests.splice(index, 1);
            res.json({ type: 'heartbeat' });
        }
    }, 30000);
});

// When an event occurs
function broadcastEvent(event) {
    pendingRequests.forEach(res => {
        res.json(event);
    });
    pendingRequests = [];
}

// Server (Node.js with ws library)
const WebSocket = require('ws');
const wss = new WebSocket.Server({ port: 8080 });

wss.on('connection', (ws) => {
    console.log('Client connected');
    
    // Send initial data
    ws.send(JSON.stringify({
        type: 'INIT',
        data: getInitialData()
    }));
    
    // Handle messages
    ws.on('message', (message) => {
        const msg = JSON.parse(message);
        
        if (msg.type === 'SUBSCRIBE') {
            subscribeToSymbol(ws, msg.symbol);
        }
    });
    
    ws.on('close', () => {
        console.log('Client disconnected');
        unsubscribeAll(ws);
    });
});

// Stream live data
function streamPriceUpdate(symbol, price) {
    wss.clients.forEach((client) => {
        if (client.readyState === WebSocket.OPEN) {
            if (isSubscribed(client, symbol)) {
                client.send(JSON.stringify({
                    type: 'PRICE_UPDATE',
                    symbol: symbol,
                    price: price,
                    timestamp: Date.now()
                }));
            }
        }
    });
}

const ws = new WebSocket('ws://localhost:8080');

ws.onopen = () => {
    console.log('Connected to server');
    
    // Subscribe to symbols
    ws.send(JSON.stringify({
        type: 'SUBSCRIBE',
        symbol: 'AAPL'
    }));
};

ws.onmessage = (event) => {
    const message = JSON.parse(event.data);
    
    switch (message.type) {
        case 'INIT':
            initializeChart(message.data);
            break;
        case 'PRICE_UPDATE':
            updateChart(message.symbol, message.price);
            break;
    }
};

ws.onerror = (error) => {
    console.error('WebSocket error:', error);
};

ws.onclose = () => {
    console.log('Disconnected, attempting reconnect...');
    setTimeout(connectWebSocket, 1000);
};

import io.vertx.core.Vertx;
import io.vertx.core.http.HttpServer;
import io.vertx.core.http.ServerWebSocket;
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.ITopic;
import com.hazelcast.core.Message;
import com.hazelcast.core.MessageListener;
import java.util.concurrent.ConcurrentHashMap;
import java.util.Set;

public class MarketDataServer {
    private final Vertx vertx;
    private final HazelcastInstance hazelcast;
    private final ConcurrentHashMap<String, Set<ServerWebSocket>> subscriptions;
    
    public MarketDataServer() {
        this.vertx = Vertx.vertx();
        this.hazelcast = Hazelcast.newHazelcastInstance();
        this.subscriptions = new ConcurrentHashMap<>();
        
        // Subscribe to market data topic
        ITopic<MarketData> topic = hazelcast.getTopic("market-data");
        topic.addMessageListener(new MessageListener<MarketData>() {
            public void onMessage(Message<MarketData> message) {
                broadcastToSubscribers(message.getMessageObject());
            }
        });
    }
    
    public void start() {
        HttpServer server = vertx.createHttpServer();
        
        server.webSocketHandler(ws -> {
            String path = ws.path();
            
            if (path.startsWith("/stream/")) {
                String symbol = path.substring(8);
                handleWebSocketConnection(ws, symbol);
            } else {
                ws.reject();
            }
        });
        
        // Long polling fallback
        server.requestHandler(req -> {
            if (req.path().startsWith("/poll/")) {
                String symbol = req.path().substring(6);
                handleLongPolling(req, symbol);
            }
        });
        
        server.listen(8080, result -> {
            if (result.succeeded()) {
                System.out.println("Market data server started on port 8080");
            }
        });
    }
    
    private void handleWebSocketConnection(ServerWebSocket ws, String symbol) {
        subscriptions.computeIfAbsent(symbol, k -> ConcurrentHashMap.newKeySet())
                     .add(ws);
        
        ws.closeHandler(v -> {
            Set<ServerWebSocket> sockets = subscriptions.get(symbol);
            if (sockets != null) {
                sockets.remove(ws);
            }
        });
        
        // Send initial snapshot from Hazelcast cache
        IMap<String, MarketData> cache = hazelcast.getMap("market-snapshot");
        MarketData data = cache.get(symbol);
        if (data != null) {
            ws.writeTextMessage(data.toJson());
        }
    }
    
    private void handleLongPolling(HttpServerRequest req, String symbol) {
        String lastEventId = req.getParam("lastEventId");
        
        // Hold request until data available or timeout
        long timerId = vertx.setTimer(30000, id -> {
            req.response()
               .putHeader("Content-Type", "application/json")
               .end("{\"type\":\"heartbeat\"}");
        });
        
        // Register one-time listener
        subscriptions.computeIfAbsent(symbol + ":poll", 
            k -> ConcurrentHashMap.newKeySet())
            .add(new PollHandler(req, timerId));
    }
    
    private void broadcastToSubscribers(MarketData data) {
        String symbol = data.getSymbol();
        
        // WebSocket subscribers
        Set<ServerWebSocket> sockets = subscriptions.get(symbol);
        if (sockets != null) {
            String json = data.toJson();
            sockets.forEach(ws -> {
                if (!ws.isClosed()) {
                    ws.writeTextMessage(json);
                }
            });
        }
        
        // Update Hazelcast cache for new subscribers
        IMap<String, MarketData> cache = hazelcast.getMap("market-snapshot");
        cache.put(symbol, data);
    }
    
    public static void main(String[] args) {
        new MarketDataServer().start();
    }
}

public class MarketDataPublisher {
    private final HazelcastInstance hazelcast;
    
    public void publishUpdate(String symbol, double price, long volume) {
        MarketData data = new MarketData(symbol, price, volume, 
                                         System.currentTimeMillis());
        
        // Publish to topic - all Vert.x instances receive it
        ITopic<MarketData> topic = hazelcast.getTopic("market-data");
        topic.publish(data);
    }
}

syntax = "proto3";

package customer;

message Customer {
    int32 customer_id = 1;
    string name = 2;
    double balance = 3;
}

service CustomerService {
    rpc GetCustomer(GetCustomerRequest) returns (Customer);
    rpc UpdateBalance(UpdateBalanceRequest) returns (UpdateBalanceResponse);
    rpc ListCustomers(ListCustomersRequest) returns (CustomerList);
}

message GetCustomerRequest {
    int32 customer_id = 1;
}

message UpdateBalanceRequest {
    int32 customer_id = 1;
    double new_balance = 2;
}

message UpdateBalanceResponse {
    bool success = 1;
}

message ListCustomersRequest {}

message CustomerList {
    repeated Customer customers = 1;
}

import grpc
from concurrent import futures
import customer_pb2
import customer_pb2_grpc

class CustomerServicer(customer_pb2_grpc.CustomerServiceServicer):
    
    def GetCustomer(self, request, context):
        return customer_pb2.Customer(
            customer_id=request.customer_id,
            name="John Doe",
            balance=5000.00
        )
    
    def UpdateBalance(self, request, context):
        print(f"Updating balance for {request.customer_id} " +
              f"to {request.new_balance}")
        return customer_pb2.UpdateBalanceResponse(success=True)
    
    def ListCustomers(self, request, context):
        customers = [
            customer_pb2.Customer(customer_id=1, name="Alice", balance=1000),
            customer_pb2.Customer(customer_id=2, name="Bob", balance=2000),
        ]
        return customer_pb2.CustomerList(customers=customers)

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    customer_pb2_grpc.add_CustomerServiceServicer_to_server(
        CustomerServicer(), server)
    server.add_insecure_port('[::]:50051')
    server.start()
    print("Server started on port 50051")
    server.wait_for_termination()

if __name__ == '__main__':
    serve()

// Lambda function (Node.js)
exports.handler = async (event) => {
    const customerId = event.queryStringParameters.customerId;
    
    // Query DynamoDB
    const AWS = require('aws-sdk');
    const dynamodb = new AWS.DynamoDB.DocumentClient();
    
    const result = await dynamodb.get({
        TableName: 'Customers',
        Key: { customerId: customerId }
    }).promise();
    
    if (result.Item) {
        return {
            statusCode: 200,
            body: JSON.stringify(result.Item)
        };
    } else {
        return {
            statusCode: 404,
            body: JSON.stringify({ error: 'Customer not found' })
        };
    }
};

{
  "Comment": "Order processing workflow",
  "StartAt": "ValidateOrder",
  "States": {
    "ValidateOrder": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123456789012:function:ValidateOrder",
      "Next": "CheckInventory"
    },
    "CheckInventory": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123456789012:function:CheckInventory",
      "Next": "ChargeCustomer"
    },
    "ChargeCustomer": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123456789012:function:ChargeCustomer",
      "Catch": [{
        "ErrorEquals": ["PaymentError"],
        "Next": "PaymentFailed"
      }],
      "Next": "ShipOrder"
    },
    "ShipOrder": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123456789012:function:ShipOrder",
      "End": true
    },
    "PaymentFailed": {
      "Type": "Fail",
      "Cause": "Payment processing failed"
    }
  }
}

syntax = "proto3";

package customer;

service CustomerService {
    rpc GetCustomer(GetCustomerRequest) returns (Customer);
    rpc UpdateCustomer(Customer) returns (UpdateResponse);
    rpc StreamOrders(StreamOrdersRequest) returns (stream Order);
    rpc BidirectionalChat(stream ChatMessage) returns (stream ChatMessage);
}

message Customer {
    int32 customer_id = 1;
    string name = 2;
    double balance = 3;
}

message GetCustomerRequest {
    int32 customer_id = 1;
}

message UpdateResponse {
    bool success = 1;
    string message = 2;
}

message StreamOrdersRequest {
    int32 customer_id = 1;
}

message Order {
    int32 order_id = 1;
    double amount = 2;
    string status = 3;
}

message ChatMessage {
    string user = 1;
    string message = 2;
    int64 timestamp = 3;
}

package main

import (
    "context"
    "fmt"
    "log"
    "net"
    "time"
    
    "google.golang.org/grpc"
    pb "example.com/customer"
)

type server struct {
    pb.UnimplementedCustomerServiceServer
}

func (s *server) GetCustomer(ctx context.Context, req *pb.GetCustomerRequest) (*pb.Customer, error) {
    return &pb.Customer{
        CustomerId: req.CustomerId,
        Name:       "John Doe",
        Balance:    5000.00,
    }, nil
}

func (s *server) UpdateCustomer(ctx context.Context, customer *pb.Customer) (*pb.UpdateResponse, error) {
    log.Printf("Updating customer %d", customer.CustomerId)
    
    return &pb.UpdateResponse{
        Success: true,
        Message: "Customer updated successfully",
    }, nil
}

func (s *server) StreamOrders(req *pb.StreamOrdersRequest, stream pb.CustomerService_StreamOrdersServer) error {
    orders := []*pb.Order{
        {OrderId: 1, Amount: 99.99, Status: "shipped"},
        {OrderId: 2, Amount: 149.50, Status: "processing"},
        {OrderId: 3, Amount: 75.25, Status: "delivered"},
    }
    
    for _, order := range orders {
        if err := stream.Send(order); err != nil {
            return err
        }
        time.Sleep(time.Second)  // Simulate delay
    }
    
    return nil
}

func (s *server) BidirectionalChat(stream pb.CustomerService_BidirectionalChatServer) error {
    for {
        msg, err := stream.Recv()
        if err != nil {
            return err
        }
        
        log.Printf("Received: %s from %s", msg.Message, msg.User)
        
        // Echo back with server prefix
        response := &pb.ChatMessage{
            User:      "Server",
            Message:   fmt.Sprintf("Echo: %s", msg.Message),
            Timestamp: time.Now().Unix(),
        }
        
        if err := stream.Send(response); err != nil {
            return err
        }
    }
}

func main() {
    lis, err := net.Listen("tcp", ":50051")
    if err != nil {
        log.Fatalf("Failed to listen: %v", err)
    }
    
    s := grpc.NewServer()
    pb.RegisterCustomerServiceServer(s, &server{})
    
    log.Println("Server listening on :50051")
    if err := s.Serve(lis); err != nil {
        log.Fatalf("Failed to serve: %v", err)
    }
}

package main

import (
    "context"
    "io"
    "log"
    "time"
    
    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials/insecure"
    pb "example.com/customer"
)

func main() {
    conn, err := grpc.Dial("localhost:50051", 
        grpc.WithTransportCredentials(insecure.NewCredentials()))
    if err != nil {
        log.Fatalf("Failed to connect: %v", err)
    }
    defer conn.Close()
    
    client := pb.NewCustomerServiceClient(conn)
    ctx := context.Background()
    
    // Unary call
    customer, err := client.GetCustomer(ctx, &pb.GetCustomerRequest{
        CustomerId: 12345,
    })
    if err != nil {
        log.Fatalf("GetCustomer failed: %v", err)
    }
    log.Printf("Customer: %v", customer)
    
    // Server streaming
    stream, err := client.StreamOrders(ctx, &pb.StreamOrdersRequest{
        CustomerId: 12345,
    })
    if err != nil {
        log.Fatalf("StreamOrders failed: %v", err)
    }
    
    for {
        order, err := stream.Recv()
        if err == io.EOF {
            break
        }
        if err != nil {
            log.Fatalf("Receive error: %v", err)
        }
        log.Printf("Order: %v", order)
    }
}

apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: grpc-service
spec:
  host: grpc-service
  trafficPolicy:
    connectionPool:
      http:
        http2MaxRequests: 100
        maxRequestsPerConnection: 10  # Force connection rotation
    loadBalancer:
      simple: LEAST_REQUEST  # Better than ROUND_ROBIN
    outlierDetection:
      consecutiveErrors: 5
      interval: 30s
      baseEjectionTime: 30s

from kafka import KafkaProducer, KafkaConsumer
import json

# Producer
producer = KafkaProducer(
    bootstrap_servers='localhost:9092',
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

order = {
    'order_id': '12345',
    'customer_id': '67890',
    'amount': 99.99,
    'timestamp': time.time()
}

producer.send('orders', value=order)
producer.flush()

# Consumer
consumer = KafkaConsumer(
    'orders',
    bootstrap_servers='localhost:9092',
    auto_offset_reset='earliest',
    value_deserializer=lambda m: json.loads(m.decode('utf-8')),
    group_id='order-processors'
)

for message in consumer:
    order = message.value
    print(f"Processing order: {order['order_id']}")
    # Process order

from mcp.server import Server
import mcp.types as types
from typing import Any
import asyncio

class DailyMinutesServer:
    def __init__(self):
        self.server = Server("daily-minutes")
        self.setup_handlers()
        
    def setup_handlers(self):
        @self.server.list_tools()
        async def handle_list_tools() -> list[types.Tool]:
            return [
                types.Tool(
                    name="get_emails",
                    description="Fetch recent emails from inbox",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "hours": {
                                "type": "number",
                                "description": "Hours to look back"
                            },
                            "limit": {
                                "type": "number", 
                                "description": "Max emails to fetch"
                            }
                        }
                    }
                ),
                types.Tool(
                    name="get_hackernews",
                    description="Fetch top Hacker News stories",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "limit": {
                                "type": "number",
                                "description": "Number of stories"
                            }
                        }
                    }
                ),
                types.Tool(
                    name="get_rss_feeds",
                    description="Fetch latest RSS feed items",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "feed_urls": {
                                "type": "array",
                                "items": {"type": "string"}
                            }
                        }
                    }
                ),
                types.Tool(
                    name="get_weather",
                    description="Get current weather forecast",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "location": {"type": "string"}
                        }
                    }
                )
            ]
        
        @self.server.call_tool()
        async def handle_call_tool(
            name: str, 
            arguments: dict[str, Any]
        ) -> list[types.TextContent]:
            if name == "get_emails":
                result = await email_connector.fetch_recent(
                    hours=arguments.get("hours", 24),
                    limit=arguments.get("limit", 10)
                )
            elif name == "get_hackernews":
                result = await hn_connector.fetch_top_stories(
                    limit=arguments.get("limit", 10)
                )
            elif name == "get_rss_feeds":
                result = await rss_connector.fetch_feeds(
                    feed_urls=arguments["feed_urls"]
                )
            elif name == "get_weather":
                result = await weather_connector.get_forecast(
                    location=arguments["location"]
                )
            else:
                raise ValueError(f"Unknown tool: {name}")
            
            return [types.TextContent(
                type="text",
                text=json.dumps(result, indent=2)
            )]

import aiohttp
from typing import List, Dict

class HackerNewsConnector:
    BASE_URL = "https://hacker-news.firebaseio.com/v0"
    
    async def fetch_top_stories(self, limit: int = 10) -> List[Dict]:
        async with aiohttp.ClientSession() as session:
            # Get top story IDs
            async with session.get(f"{self.BASE_URL}/topstories.json") as resp:
                story_ids = await resp.json()
            
            # Fetch details for top N stories
            stories = []
            for story_id in story_ids[:limit]:
                async with session.get(
                    f"{self.BASE_URL}/item/{story_id}.json"
                ) as resp:
                    story = await resp.json()
                    stories.append({
                        "title": story.get("title"),
                        "url": story.get("url"),
                        "score": story.get("score"),
                        "by": story.get("by"),
                        "time": story.get("time")
                    })
            
            return stories

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "get_emails",
    "arguments": {
      "hours": 12,
      "limit": 5
    }
  }
}

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "content": [
      {
        "type": "text",
        "text": "[{\"from\": \"john@example.com\", \"subject\": \"Q4 Review\", ...}]"
      }
    ]
  }
}

from a2a import Agent, Capability

class ResearchAgent(Agent):
    def __init__(self):
        super().__init__(
            agent_id="research-agent-01",
            name="Research Agent",
            description="Performs web research and summarization"
        )
        
        # Register capabilities
        self.register_capability(
            Capability(
                name="web_search",
                description="Search the web for information",
                input_schema={
                    "type": "object",
                    "properties": {
                        "query": {"type": "string"},
                        "max_results": {"type": "integer", "default": 10}
                    },
                    "required": ["query"]
                },
                output_schema={
                    "type": "object",
                    "properties": {
                        "results": {
                            "type": "array",
                            "items": {
                                "type": "object",
                                "properties": {
                                    "title": {"type": "string"},
                                    "url": {"type": "string"},
                                    "snippet": {"type": "string"}
                                }
                            }
                        }
                    }
                }
            )
        )
    
    async def handle_request(self, capability: str, params: dict):
        if capability == "web_search":
            return await self.perform_web_search(
                query=params["query"],
                max_results=params.get("max_results", 10)
            )
    
    async def perform_web_search(self, query: str, max_results: int):
        # Actual search implementation
        results = await search_engine.search(query, limit=max_results)
        return {"results": results}

class CoordinatorAgent(Agent):
    def __init__(self):
        super().__init__(
            agent_id="coordinator-01",
            name="Coordinator Agent"
        )
        self.directory = AgentDirectory()
    
    async def research_topic(self, topic: str):
        # Discover agents with web_search capability
        agents = await self.directory.find_agents_with_capability("web_search")
        
        if not agents:
            raise Exception("No research agents available")
        
        # Select an agent (load balancing, availability, etc.)
        research_agent = agents[0]
        
        # Invoke the capability via JSON-RPC
        result = await research_agent.invoke(
            capability="web_search",
            params={
                "query": topic,
                "max_results": 20
            }
        )
        
        return result

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "directory.find_agents",
  "params": {
    "capability": "web_search",
    "filters": {
      "availability": "online"
    }
  }
}

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "agents": [
      {
        "agent_id": "research-agent-01",
        "name": "Research Agent",
        "endpoint": "http://agent-service:8080/rpc",
        "capabilities": ["web_search"],
        "metadata": {
          "load": 0.3,
          "response_time_ms": 150
        }
      }
    ]
  }
}

HTTP Client ? gRPC-Gateway ? gRPC Server
     ?              ?            ?
REST Request   Proto Message   gRPC Service

POST /v1/users
Authorization: Bearer abc123
X-Request-ID: req-456
X-User-Role: admin
Content-Type: application/json

metadata.MD{
    "authorization": []string{"Bearer abc123"},
    "request-id":    []string{"req-456"}, 
    "user-role":     []string{"admin"},
}

X-Request-ID: req-456
X-Processing-Time: 150ms
X-Server-Version: v1.2.0

// Service A approach
func (s *ServiceA) CreateUser(ctx context.Context, req *pb.CreateUserRequest) (*pb.User, error) {
    md, _ := metadata.FromIncomingContext(ctx)
    authHeader := md.Get("authorization")
    userID := md.Get("x-user-id")
    // ... custom mapping logic
}

// Service B approach  
func (s *ServiceB) GetOrder(ctx context.Context, req *pb.GetOrderRequest) (*pb.Order, error) {
    // Different header names, different extraction logic
    md, _ := metadata.FromIncomingContext(ctx)
    auth := md.Get("auth")  // Different from Service A!
    requestID := md.Get("request_id")  // Different format!
}

// ANTI-PATTERN: Storing large objects in context
type UserContext struct {
    User        *User           // Large user object
    Permissions []Permission    // Array of permissions  
    Preferences *UserPrefs      // User preferences
    AuditLog    []AuditEntry   // Historical data
}

func StoreUserInContext(ctx context.Context, user *UserContext) context.Context {
    return context.WithValue(ctx, "user", user)  // BAD: Large object in context
}

// GOOD: Store only identifiers in context  
func StoreUserIDInContext(ctx context.Context, userID string) context.Context {
    return context.WithValue(ctx, "user_id", userID)  // Small string only
}

// Fetch data when needed from database/cache
func GetUserFromContext(ctx context.Context) (*User, error) {
    userID := ctx.Value("user_id").(string)
    return userService.GetUser(userID)  // Fetch from datastore
}

// gRPC: Set headers via metadata
grpc.SendHeader(ctx, metadata.New(map[string]string{
    "x-server-version": "v1.2.0",
}))

// HTTP: Set headers on ResponseWriter  
w.Header().Set("X-Server-Version", "v1.2.0")

// gRPC-Gateway: Headers must be set in specific metadata format
grpc.SetHeader(ctx, metadata.New(map[string]string{
    "grpc-metadata-x-server-version": "v1.2.0",  // Prefix required
}))

HTTP Request ? Gateway Middleware ? gRPC Interceptor ? Service
     ?              ?                    ?              ?
HTTP Headers ? Metadata Annotation ? Context Metadata ? Business Logic
                                                         ?
HTTP Response ? Response Modifier ? Header Metadata ? Service Response

mapper := headermapper.NewBuilder().
    // Authentication headers
    AddIncomingMapping("Authorization", "authorization").WithRequired(true).
    AddIncomingMapping("X-API-Key", "api-key").
    
    // Request tracking (bidirectional)  
    AddBidirectionalMapping("X-Request-ID", "request-id").
    AddBidirectionalMapping("X-Trace-ID", "trace-id").
    
    // Response headers
    AddOutgoingMapping("processing-time", "X-Processing-Time").
    AddOutgoingMapping("server-version", "X-Server-Version").
    
    // Transformations
    AddIncomingMapping("Authorization", "auth-token").
    WithTransform(headermapper.ChainTransforms(
        headermapper.TrimSpace,
        headermapper.RemovePrefix("Bearer "),
    )).
    
    Build()

POST /v1/users HTTP/1.1
Authorization: Bearer abc123
X-Request-ID: req-456
X-User-Role: admin
Content-Type: application/json

func (hm *HeaderMapper) MetadataAnnotator() func(context.Context, *http.Request) metadata.MD {
    return func(ctx context.Context, req *http.Request) metadata.MD {
        md := metadata.New(map[string]string{})
        
        for _, mapping := range hm.config.Mappings {
            if mapping.Direction == Outgoing {
                continue  // Skip outgoing-only mappings
            }
            
            headerValue := req.Header.Get(mapping.HTTPHeader)
            if headerValue != "" {
                // Apply transformations if configured
                if mapping.Transform != nil {
                    headerValue = mapping.Transform(headerValue)
                }
                md.Set(mapping.GRPCMetadata, headerValue)
            }
        }
        return md
    }
}

metadata.MD{
    "authorization": []string{"Bearer abc123"},
    "auth-token":    []string{"abc123"},        // Transformed  
    "request-id":    []string{"req-456"},
    "user-role":     []string{"admin"},
}

func (hm *HeaderMapper) UnaryServerInterceptor() grpc.UnaryServerInterceptor {
    return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
        // Context already contains mapped metadata from Step 1
        
        // Call the actual service method
        resp, err := handler(ctx, req)
        
        // Response headers are handled by ResponseModifier
        return resp, err
    }
}

func (s *UserService) CreateUser(ctx context.Context, req *pb.CreateUserRequest) (*pb.User, error) {
    md, _ := metadata.FromIncomingContext(ctx)
    
    // Headers are consistently available
    authToken := getFirstValue(md, "auth-token")      // "abc123" (transformed)
    requestID := getFirstValue(md, "request-id")      // "req-456"  
    userRole := getFirstValue(md, "user-role")        // "admin"
    
    // Set response headers
    grpc.SetHeader(ctx, metadata.New(map[string]string{
        "processing-time": "150",
        "server-version": "v1.2.0",  
        "request-id": requestID,     // Echo back request ID
    }))
    
    return &pb.User{...}, nil
}

func (hm *HeaderMapper) ResponseModifier() func(context.Context, http.ResponseWriter, proto.Message) error {
    return func(ctx context.Context, w http.ResponseWriter, msg proto.Message) error {
        md, ok := runtime.ServerMetadataFromContext(ctx)
        if !ok {
            return nil
        }
        
        for _, mapping := range hm.config.Mappings {
            if mapping.Direction == Incoming {
                continue  // Skip incoming-only mappings  
            }
            
            values := md.HeaderMD.Get(mapping.GRPCMetadata)
            if len(values) > 0 {
                headerValue := values[0]
                
                // Apply transformations
                if mapping.Transform != nil {
                    headerValue = mapping.Transform(headerValue)  
                }
                
                w.Header().Set(mapping.HTTPHeader, headerValue)
            }
        }
        return nil
    }
}

HTTP/1.1 200 OK
X-Request-ID: req-456
X-Processing-Time: 150ms  
X-Server-Version: v1.2.0
Content-Type: application/json

{"user": {...}}

// Extract JWT tokens
AddIncomingMapping("Authorization", "jwt-token").
WithTransform(headermapper.ChainTransforms(
    headermapper.TrimSpace,
    headermapper.RemovePrefix("Bearer "),
    headermapper.Truncate(100),  // Prevent large tokens
))

// Sanitize user agents
AddIncomingMapping("User-Agent", "client-info").  
WithTransform(headermapper.RegexReplace(`\d+\.\d+(\.\d+)*`, "x.x.x"))

// Format timestamps
AddOutgoingMapping("response-time", "X-Response-Time").
WithTransform(headermapper.AddSuffix("ms"))

# header-mapping.yaml
mappings:
  - http_header: "Authorization"
    grpc_metadata: "authorization" 
    direction: 0  # Incoming
    required: true
    
  - http_header: "X-Request-ID"
    grpc_metadata: "request-id"
    direction: 2  # Bidirectional
    default_value: "auto-generated"

skip_paths:
  - "/health"
  - "/metrics"
  
debug: false

config, err := headermapper.LoadConfigFromFile("header-mapping.yaml")
if err != nil {
    log.Fatal("Failed to load config:", err)
}

mapper := headermapper.NewHeaderMapper(config)

mapper := headermapper.NewBuilder().
    AddIncomingMapping("Authorization", "authorization").
    SkipPaths("/health", "/metrics", "/debug").  // No auth required
    Build()

package main

import (
    "github.com/your-org/grpc-header-mapper/headermapper"
    "github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
)

func main() {
    // Create header mapper
    mapper := headermapper.NewBuilder().
        AddIncomingMapping("Authorization", "authorization").
        AddBidirectionalMapping("X-Request-ID", "request-id").
        Build()
    
    // Configure gRPC server
    grpcServer := grpc.NewServer(
        grpc.UnaryInterceptor(mapper.UnaryServerInterceptor()),
    )
    
    // Configure HTTP gateway
    mux := headermapper.CreateGatewayMux(mapper)
    
    // Register services...
}

func createProductionMapper() *headermapper.HeaderMapper {
    return headermapper.NewBuilder().
        // Authentication
        AddIncomingMapping("Authorization", "authorization").WithRequired(true).
        AddIncomingMapping("X-API-Key", "api-key").
        
        // Request correlation
        AddBidirectionalMapping("X-Request-ID", "request-id").
        AddBidirectionalMapping("X-Correlation-ID", "correlation-id"). 
        AddBidirectionalMapping("X-Trace-ID", "trace-id").
        
        // Client information
        AddIncomingMapping("User-Agent", "user-agent").
        AddIncomingMapping("X-Client-Version", "client-version").
        
        // Response headers
        AddOutgoingMapping("processing-time-ms", "X-Processing-Time").
        AddOutgoingMapping("server-version", "X-Server-Version").
        AddOutgoingMapping("rate-limit-remaining", "X-RateLimit-Remaining").
        
        // Security headers
        AddOutgoingMapping("content-security-policy", "Content-Security-Policy").
        WithDefault("default-src 'self'").
        
        // Skip system endpoints
        SkipPaths("/health", "/metrics", "/debug", "/admin").
        
        // Production settings
        Debug(false).
        OverwriteExisting(true).
        Build()
}

// Before: Inconsistent, potentially large context values
ctx = context.WithValue(ctx, "user", largeUserObject)      // BAD
ctx = context.WithValue(ctx, "permissions", permissionList) // BAD

// After: Consistent, minimal context usage  
// Headers extracted to standard metadata, large objects fetched on-demand
func GetUserFromContext(ctx context.Context) (*User, error) {
    userID := getMetadata(ctx, "user-id")
    return userCache.Get(userID)  // Cached lookup
}

// All services automatically have request correlation
func (s *AnyService) AnyMethod(ctx context.Context, req *AnyRequest) (*AnyResponse, error) {
    requestID := getMetadata(ctx, "request-id")  // Always available
    log.WithField("request_id", requestID).Info("Processing request")
    
    // Business logic...
    
    return response, nil
}

func TestServiceWithHeaders(t *testing.T) {
    // Headers work consistently across all services
    client := pb.NewUserServiceClient(conn)
    
    ctx := metadata.NewOutgoingContext(context.Background(), metadata.New(map[string]string{
        "authorization": "Bearer test-token",
        "request-id":    "test-req-123",
    }))
    
    resp, err := client.CreateUser(ctx, &pb.CreateUserRequest{...})
    
    // Response headers are consistently available
    md, _ := metadata.FromIncomingContext(ctx)
    requestID := getMetadata(md, "request-id")  // "test-req-123"
}

mapper := headermapper.NewBuilder().
    AddIncomingMapping("Authorization", "authorization").
    WithTransform(headermapper.ChainTransforms(
        headermapper.TrimSpace,
        headermapper.Truncate(512),  // Prevent oversized headers
        validateJWTFormat,           // Custom validation
    )).
    Build()

func validateJWTFormat(token string) string {
    if !strings.HasPrefix(token, "Bearer ") {
        return "invalid"  // Reject malformed tokens
    }
    return token
}

AddIncomingMapping("Authorization", "authorization").
WithTransform(headermapper.MaskSensitive(4)).  // Show first/last 4 chars

AddOutgoingMapping("rate-limit-remaining", "X-RateLimit-Remaining").
AddOutgoingMapping("rate-limit-reset", "X-RateLimit-Reset").

mapper := headermapper.NewBuilder().
    Debug(true).  // Enable detailed logging
    Build()

mapper.SetLogger(customLogger)  // Use your logging framework

[DEBUG] [HeaderMapper] Mapped incoming headers: map[authorization:[Bearer abc123] request-id:[req-456]]
[DEBUG] [HeaderMapper] Mapped outgoing headers to response  

stats := mapper.GetStats()
prometheus.IncomingHeadersMappedCounter.Add(stats.IncomingMappings)
prometheus.OutgoingHeadersMappedCounter.Add(stats.OutgoingMappings)
prometheus.MappingErrorsCounter.Add(stats.FailedMappings)

{
  "type": "https://example.com/docs/errors/validation-failed",
  "title": "Validation Failed",
  "status": 400,
  "detail": "The request body failed validation.",
  "instance": "/v1/todos",
  "invalid_params": [
    {
      "field": "title",
      "reason": "must not be empty"
    }
  ]
}

syntax = "proto3";

package todo.v1;

import "google/api/annotations.proto";
import "google/api/field_behavior.proto";
import "google/api/resource.proto";
import "google/protobuf/timestamp.proto";
import "google/protobuf/field_mask.proto";
import "buf/validate/validate.proto";

option go_package = "github.com/bhatti/todo-api-errors/api/proto/todo/v1;todo";

// TodoService provides task management operations
service TodoService {
  // CreateTask creates a new task
  rpc CreateTask(CreateTaskRequest) returns (Task) {
    option (google.api.http) = {
      post: "/v1/tasks"
      body: "*"
    };
  }

  // GetTask retrieves a specific task
  rpc GetTask(GetTaskRequest) returns (Task) {
    option (google.api.http) = {
      get: "/v1/{name=tasks/*}"
    };
  }

  // ListTasks retrieves all tasks
  rpc ListTasks(ListTasksRequest) returns (ListTasksResponse) {
    option (google.api.http) = {
      get: "/v1/tasks"
    };
  }

  // UpdateTask updates an existing task
  rpc UpdateTask(UpdateTaskRequest) returns (Task) {
    option (google.api.http) = {
      patch: "/v1/{task.name=tasks/*}"
      body: "task"
    };
  }

  // DeleteTask removes a task
  rpc DeleteTask(DeleteTaskRequest) returns (DeleteTaskResponse) {
    option (google.api.http) = {
      delete: "/v1/{name=tasks/*}"
    };
  }

  // BatchCreateTasks creates multiple tasks at once
  rpc BatchCreateTasks(BatchCreateTasksRequest) returns (BatchCreateTasksResponse) {
    option (google.api.http) = {
      post: "/v1/tasks:batchCreate"
      body: "*"
    };
  }
}

// Task represents a TODO item
message Task {
  option (google.api.resource) = {
    type: "todo.example.com/Task"
    pattern: "tasks/{task}"
    singular: "task"
    plural: "tasks"
  };

  // Resource name of the task
  string name = 1 [
    (google.api.field_behavior) = IDENTIFIER,
    (google.api.field_behavior) = OUTPUT_ONLY
  ];

  // Task title
  string title = 2 [
    (google.api.field_behavior) = REQUIRED,
    (buf.validate.field).string = {
      min_len: 1
      max_len: 200
    }
  ];

  // Task description
  string description = 3 [
    (google.api.field_behavior) = OPTIONAL,
    (buf.validate.field).string = {
      max_len: 1000
    }
  ];

  // Task status
  Status status = 4 [
    (google.api.field_behavior) = REQUIRED
  ];

  // Task priority
  Priority priority = 5 [
    (google.api.field_behavior) = OPTIONAL
  ];

  // Due date for the task
  google.protobuf.Timestamp due_date = 6 [
    (google.api.field_behavior) = OPTIONAL,
    (buf.validate.field).timestamp = {
      gt_now: true
    }
  ];

  // Task creation time
  google.protobuf.Timestamp create_time = 7 [
    (google.api.field_behavior) = OUTPUT_ONLY
  ];

  // Task last update time
  google.protobuf.Timestamp update_time = 8 [
    (google.api.field_behavior) = OUTPUT_ONLY
  ];

  // User who created the task
  string created_by = 9 [
    (google.api.field_behavior) = OUTPUT_ONLY
  ];

  // Tags associated with the task
  repeated string tags = 10 [
    (buf.validate.field).repeated = {
      max_items: 10
      items: {
        string: {
          pattern: "^[a-z0-9-]+$"
          max_len: 50
        }
      }
    }
  ];
}

// Task status enumeration
enum Status {
  STATUS_UNSPECIFIED = 0;
  STATUS_PENDING = 1;
  STATUS_IN_PROGRESS = 2;
  STATUS_COMPLETED = 3;
  STATUS_CANCELLED = 4;
}

// Task priority enumeration
enum Priority {
  PRIORITY_UNSPECIFIED = 0;
  PRIORITY_LOW = 1;
  PRIORITY_MEDIUM = 2;
  PRIORITY_HIGH = 3;
  PRIORITY_CRITICAL = 4;
}

// CreateTaskRequest message
message CreateTaskRequest {
  // Task to create
  Task task = 1 [
    (google.api.field_behavior) = REQUIRED,
    (buf.validate.field).required = true
  ];
}

// GetTaskRequest message
message GetTaskRequest {
  // Resource name of the task
  string name = 1 [
    (google.api.field_behavior) = REQUIRED,
    (google.api.resource_reference) = {
      type: "todo.example.com/Task"
    },
    (buf.validate.field).string = {
      pattern: "^tasks/[a-zA-Z0-9-]+$"
    }
  ];
}

// ListTasksRequest message
message ListTasksRequest {
  // Maximum number of tasks to return
  int32 page_size = 1 [
    (buf.validate.field).int32 = {
      gte: 0
      lte: 1000
    }
  ];

  // Page token for pagination
  string page_token = 2;

  // Filter expression
  string filter = 3;

  // Order by expression
  string order_by = 4;
}

// ListTasksResponse message
message ListTasksResponse {
  // List of tasks
  repeated Task tasks = 1;

  // Token for next page
  string next_page_token = 2;

  // Total number of tasks
  int32 total_size = 3;
}

// UpdateTaskRequest message
message UpdateTaskRequest {
  // Task to update
  Task task = 1 [
    (google.api.field_behavior) = REQUIRED,
    (buf.validate.field).required = true
  ];

  // Fields to update
  google.protobuf.FieldMask update_mask = 2 [
    (google.api.field_behavior) = REQUIRED,
    (buf.validate.field).required = true
  ];
}

// DeleteTaskRequest message
message DeleteTaskRequest {
  // Resource name of the task
  string name = 1 [
    (google.api.field_behavior) = REQUIRED,
    (google.api.resource_reference) = {
      type: "todo.example.com/Task"
    }
  ];
}

// DeleteTaskResponse message
message DeleteTaskResponse {
  // Confirmation message
  string message = 1;
}

// BatchCreateTasksRequest message
message BatchCreateTasksRequest {
  // Tasks to create
  repeated CreateTaskRequest requests = 1 [
    (google.api.field_behavior) = REQUIRED,
    (buf.validate.field).repeated = {
      min_items: 1
      max_items: 100
    }
  ];
}

// BatchCreateTasksResponse message
message BatchCreateTasksResponse {
  // Created tasks
  repeated Task tasks = 1;
}

syntax = "proto3";

package errors.v1;

import "google/protobuf/timestamp.proto";
import "google/protobuf/any.proto";

option go_package = "github.com/bhatti/todo-api-errors/api/proto/errors/v1;errors";

// ErrorDetail provides a structured, machine-readable error payload.
// It is designed to be embedded in the `details` field of a `google.rpc.Status` message.
message ErrorDetail {
  // A unique, application-specific error code.
  string code = 1;
  // A short, human-readable summary of the problem type.
  string title = 2;
  // A human-readable explanation specific to this occurrence of the problem.
  string detail = 3;
  // A list of validation errors, useful for INVALID_ARGUMENT responses.
  repeated FieldViolation field_violations = 4;
  // Optional trace ID for request correlation
  string trace_id = 5;
  // Optional timestamp when the error occurred
  google.protobuf.Timestamp timestamp = 6;
  // Optional instance path where the error occurred
  string instance = 7;
  // Optional extensions for additional error context
  map<string, google.protobuf.Any> extensions = 8;
}

// Describes a single validation failure.
message FieldViolation {
  // The path to the field that failed validation, e.g., "title".
  string field = 1;
  // A developer-facing description of the validation rule that failed.
  string description = 2;
  // Application-specific error code for this validation failure
  string code = 3;
}

// AppErrorCode defines a list of standardized, application-specific error codes.
enum AppErrorCode {
  APP_ERROR_CODE_UNSPECIFIED = 0;

  // Validation failures
  VALIDATION_FAILED = 1;
  REQUIRED_FIELD = 2;
  TOO_SHORT = 3;
  TOO_LONG = 4;
  INVALID_FORMAT = 5;
  MUST_BE_FUTURE = 6;
  INVALID_VALUE = 7;
  DUPLICATE_TAG = 8;
  INVALID_TAG_FORMAT = 9;
  OVERDUE_COMPLETION = 10;
  EMPTY_BATCH = 11;
  BATCH_TOO_LARGE = 12;
  DUPLICATE_TITLE = 13;

  // Resource errors
  RESOURCE_NOT_FOUND = 1001;
  RESOURCE_CONFLICT = 1002;

  // Authentication and authorization
  AUTHENTICATION_FAILED = 2001;
  PERMISSION_DENIED = 2002;

  // Rate limiting and service availability
  RATE_LIMIT_EXCEEDED = 3001;
  SERVICE_UNAVAILABLE = 3002;

  // Internal errors
  INTERNAL_ERROR = 9001;
}

package errors

import (
	"fmt"

	errorspb "github.com/bhatti/todo-api-errors/api/proto/errors/v1"
	"google.golang.org/genproto/googleapis/rpc/errdetails"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/types/known/anypb"
	"google.golang.org/protobuf/types/known/timestamppb"
)

// AppError is our custom error type using protobuf definitions.
type AppError struct {
	GRPCCode        codes.Code
	AppCode         errorspb.AppErrorCode
	Title           string
	Detail          string
	FieldViolations []*errorspb.FieldViolation
	TraceID         string
	Instance        string
	Extensions      map[string]*anypb.Any
	CausedBy        error // For internal logging
}

func (e *AppError) Error() string {
	return fmt.Sprintf("gRPC Code: %s, App Code: %s, Title: %s, Detail: %s", e.GRPCCode, e.AppCode, e.Title, e.Detail)
}

// ToGRPCStatus converts our AppError into a gRPC status.Status.
func (e *AppError) ToGRPCStatus() *status.Status {
	st := status.New(e.GRPCCode, e.Title)

	errorDetail := &errorspb.ErrorDetail{
		Code:            e.AppCode.String(),
		Title:           e.Title,
		Detail:          e.Detail,
		FieldViolations: e.FieldViolations,
		TraceId:         e.TraceID,
		Timestamp:       timestamppb.Now(),
		Instance:        e.Instance,
		Extensions:      e.Extensions,
	}

	// For validation errors, we also attach the standard BadRequest detail
	// so that gRPC-Gateway and other standard tools can understand it.
	if e.GRPCCode == codes.InvalidArgument && len(e.FieldViolations) > 0 {
		br := &errdetails.BadRequest{}
		for _, fv := range e.FieldViolations {
			br.FieldViolations = append(br.FieldViolations, &errdetails.BadRequest_FieldViolation{
				Field:       fv.Field,
				Description: fv.Description,
			})
		}
		st, _ = st.WithDetails(br, errorDetail)
		return st
	}

	st, _ = st.WithDetails(errorDetail)
	return st
}

// Helper functions for creating common errors

func NewValidationFailed(violations []*errorspb.FieldViolation, traceID string) *AppError {
	return &AppError{
		GRPCCode:        codes.InvalidArgument,
		AppCode:         errorspb.AppErrorCode_VALIDATION_FAILED,
		Title:           "Validation Failed",
		Detail:          fmt.Sprintf("The request contains %d validation errors", len(violations)),
		FieldViolations: violations,
		TraceID:         traceID,
	}
}

func NewNotFound(resource string, id string, traceID string) *AppError {
	return &AppError{
		GRPCCode: codes.NotFound,
		AppCode:  errorspb.AppErrorCode_RESOURCE_NOT_FOUND,
		Title:    "Resource Not Found",
		Detail:   fmt.Sprintf("%s with ID '%s' was not found.", resource, id),
		TraceID:  traceID,
	}
}

func NewConflict(resource, reason string, traceID string) *AppError {
	return &AppError{
		GRPCCode: codes.AlreadyExists,
		AppCode:  errorspb.AppErrorCode_RESOURCE_CONFLICT,
		Title:    "Resource Conflict",
		Detail:   fmt.Sprintf("Conflict creating %s: %s", resource, reason),
		TraceID:  traceID,
	}
}

func NewInternal(message string, traceID string, causedBy error) *AppError {
	return &AppError{
		GRPCCode: codes.Internal,
		AppCode:  errorspb.AppErrorCode_INTERNAL_ERROR,
		Title:    "Internal Server Error",
		Detail:   message,
		TraceID:  traceID,
		CausedBy: causedBy,
	}
}

func NewPermissionDenied(resource, action string, traceID string) *AppError {
	return &AppError{
		GRPCCode: codes.PermissionDenied,
		AppCode:  errorspb.AppErrorCode_PERMISSION_DENIED,
		Title:    "Permission Denied",
		Detail:   fmt.Sprintf("You don't have permission to %s %s", action, resource),
		TraceID:  traceID,
	}
}

func NewServiceUnavailable(message string, traceID string) *AppError {
	return &AppError{
		GRPCCode: codes.Unavailable,
		AppCode:  errorspb.AppErrorCode_SERVICE_UNAVAILABLE,
		Title:    "Service Unavailable",
		Detail:   message,
		TraceID:  traceID,
	}
}

func NewRequiredField(field, message string, traceID string) *AppError {
	return &AppError{
		GRPCCode: codes.InvalidArgument,
		AppCode:  errorspb.AppErrorCode_VALIDATION_FAILED,
		Title:    "Validation Failed",
		Detail:   "The request contains validation errors",
		FieldViolations: []*errorspb.FieldViolation{
			{
				Field:       field,
				Code:        errorspb.AppErrorCode_REQUIRED_FIELD.String(),
				Description: message,
			},
		},
		TraceID: traceID,
	}
}

package validation

import (
	"errors"
	"fmt"
	"regexp"
	"strings"

	"buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go/buf/validate"
	"buf.build/go/protovalidate"
	errorspb "github.com/bhatti/todo-api-errors/api/proto/errors/v1"
	todopb "github.com/bhatti/todo-api-errors/api/proto/todo/v1"
	apperrors "github.com/bhatti/todo-api-errors/internal/errors"
	"google.golang.org/protobuf/proto"
)

var pv protovalidate.Validator

func init() {
	var err error
	pv, err = protovalidate.New()
	if err != nil {
		panic(fmt.Sprintf("failed to initialize protovalidator: %v", err))
	}
}

// ValidateRequest checks a proto message and returns an AppError with all violations.
func ValidateRequest(req proto.Message, traceID string) error {
	if err := pv.Validate(req); err != nil {
		var validationErrs *protovalidate.ValidationError
		if errors.As(err, &validationErrs) {
			var violations []*errorspb.FieldViolation
			for _, violation := range validationErrs.Violations {
				fieldPath := ""
				if violation.Proto.GetField() != nil {
					fieldPath = formatFieldPath(violation.Proto.GetField())
				}

				ruleId := violation.Proto.GetRuleId()
				message := violation.Proto.GetMessage()

				violations = append(violations, &errorspb.FieldViolation{
					Field:       fieldPath,
					Description: message,
					Code:        mapConstraintToCode(ruleId),
				})
			}
			return apperrors.NewValidationFailed(violations, traceID)
		}
		return apperrors.NewInternal("Validation failed", traceID, err)
	}
	return nil
}

// ValidateTask performs additional business logic validation
func ValidateTask(task *todopb.Task, traceID string) error {
	var violations []*errorspb.FieldViolation

	// Proto validation first
	if err := ValidateRequest(task, traceID); err != nil {
		if appErr, ok := err.(*apperrors.AppError); ok {
			violations = append(violations, appErr.FieldViolations...)
		}
	}

	// Additional business rules
	if task.Status == todopb.Status_STATUS_COMPLETED && task.DueDate != nil {
		if task.UpdateTime != nil && task.UpdateTime.AsTime().After(task.DueDate.AsTime()) {
			violations = append(violations, &errorspb.FieldViolation{
				Field:       "due_date",
				Code:        errorspb.AppErrorCode_OVERDUE_COMPLETION.String(),
				Description: "Task was completed after the due date",
			})
		}
	}

	// Validate tags format
	for i, tag := range task.Tags {
		if !isValidTag(tag) {
			violations = append(violations, &errorspb.FieldViolation{
				Field:       fmt.Sprintf("tags[%d]", i),
				Code:        errorspb.AppErrorCode_INVALID_TAG_FORMAT.String(),
				Description: fmt.Sprintf("Tag '%s' must be lowercase letters, numbers, and hyphens only", tag),
			})
		}
	}

	// Check for duplicate tags
	tagMap := make(map[string]bool)
	for i, tag := range task.Tags {
		if tagMap[tag] {
			violations = append(violations, &errorspb.FieldViolation{
				Field:       fmt.Sprintf("tags[%d]", i),
				Code:        errorspb.AppErrorCode_DUPLICATE_TAG.String(),
				Description: fmt.Sprintf("Tag '%s' appears multiple times", tag),
			})
		}
		tagMap[tag] = true
	}

	if len(violations) > 0 {
		return apperrors.NewValidationFailed(violations, traceID)
	}

	return nil
}

// ValidateBatchCreateTasks validates batch operations
func ValidateBatchCreateTasks(req *todopb.BatchCreateTasksRequest, traceID string) error {
	var violations []*errorspb.FieldViolation

	// Check batch size
	if len(req.Requests) == 0 {
		violations = append(violations, &errorspb.FieldViolation{
			Field:       "requests",
			Code:        errorspb.AppErrorCode_EMPTY_BATCH.String(),
			Description: "Batch must contain at least one task",
		})
	}

	if len(req.Requests) > 100 {
		violations = append(violations, &errorspb.FieldViolation{
			Field:       "requests",
			Code:        errorspb.AppErrorCode_BATCH_TOO_LARGE.String(),
			Description: fmt.Sprintf("Batch size %d exceeds maximum of 100", len(req.Requests)),
		})
	}

	// Validate each task
	for i, createReq := range req.Requests {
		if createReq.Task == nil {
			violations = append(violations, &errorspb.FieldViolation{
				Field:       fmt.Sprintf("requests[%d].task", i),
				Code:        errorspb.AppErrorCode_REQUIRED_FIELD.String(),
				Description: "Task is required",
			})
			continue
		}

		// Validate task
		if err := ValidateTask(createReq.Task, traceID); err != nil {
			if appErr, ok := err.(*apperrors.AppError); ok {
				for _, violation := range appErr.FieldViolations {
					violation.Field = fmt.Sprintf("requests[%d].task.%s", i, violation.Field)
					violations = append(violations, violation)
				}
			}
		}
	}

	// Check for duplicate titles
	titleMap := make(map[string][]int)
	for i, createReq := range req.Requests {
		if createReq.Task != nil && createReq.Task.Title != "" {
			titleMap[createReq.Task.Title] = append(titleMap[createReq.Task.Title], i)
		}
	}

	for title, indices := range titleMap {
		if len(indices) > 1 {
			for _, idx := range indices {
				violations = append(violations, &errorspb.FieldViolation{
					Field:       fmt.Sprintf("requests[%d].task.title", idx),
					Code:        errorspb.AppErrorCode_DUPLICATE_TITLE.String(),
					Description: fmt.Sprintf("Title '%s' is used by multiple tasks in the batch", title),
				})
			}
		}
	}

	if len(violations) > 0 {
		return apperrors.NewValidationFailed(violations, traceID)
	}

	return nil
}

// Helper functions
func formatFieldPath(fieldPath *validate.FieldPath) string {
	if fieldPath == nil {
		return ""
	}

	// Build field path from elements
	var parts []string
	for _, element := range fieldPath.GetElements() {
		if element.GetFieldName() != "" {
			parts = append(parts, element.GetFieldName())
		} else if element.GetFieldNumber() != 0 {
			parts = append(parts, fmt.Sprintf("field_%d", element.GetFieldNumber()))
		}
	}

	return strings.Join(parts, ".")
}

func mapConstraintToCode(ruleId string) string {
	switch {
	case strings.Contains(ruleId, "required"):
		return errorspb.AppErrorCode_REQUIRED_FIELD.String()
	case strings.Contains(ruleId, "min_len"):
		return errorspb.AppErrorCode_TOO_SHORT.String()
	case strings.Contains(ruleId, "max_len"):
		return errorspb.AppErrorCode_TOO_LONG.String()
	case strings.Contains(ruleId, "pattern"):
		return errorspb.AppErrorCode_INVALID_FORMAT.String()
	case strings.Contains(ruleId, "gt_now"):
		return errorspb.AppErrorCode_MUST_BE_FUTURE.String()
	case ruleId == "":
		return errorspb.AppErrorCode_VALIDATION_FAILED.String()
	default:
		return errorspb.AppErrorCode_INVALID_VALUE.String()
	}
}

var validTagPattern = regexp.MustCompile(`^[a-z0-9-]+$`)

func isValidTag(tag string) bool {
	return len(tag) <= 50 && validTagPattern.MatchString(tag)
}

package middleware

import (
	"context"
	"errors"
	"log"

	apperrors "github.com/bhatti/todo-api-errors/internal/errors"
	"google.golang.org/grpc"
	"google.golang.org/grpc/status"
)

// UnaryErrorInterceptor translates application errors into gRPC statuses.
func UnaryErrorInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
	resp, err := handler(ctx, req)
	if err == nil {
		return resp, nil
	}

	var appErr *apperrors.AppError
	if errors.As(err, &appErr) {
		if appErr.CausedBy != nil {
			log.Printf("ERROR: %s, Original cause: %v", appErr.Title, appErr.CausedBy)
		}
		return nil, appErr.ToGRPCStatus().Err()
	}

	if _, ok := status.FromError(err); ok {
		return nil, err // Already a gRPC status
	}

	log.Printf("UNEXPECTED ERROR: %v", err)
	return nil, apperrors.NewInternal("An unexpected error occurred", "", err).ToGRPCStatus().Err()
}

package middleware

import (
	"context"
	"encoding/json"
	"net/http"
	"runtime/debug"
	"time"

	errorspb "github.com/bhatti/todo-api-errors/api/proto/errors/v1"
	apperrors "github.com/bhatti/todo-api-errors/internal/errors"
	"github.com/google/uuid"
	"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
	"go.opentelemetry.io/otel/trace"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/encoding/protojson"
)

// HTTPErrorHandler handles errors for HTTP endpoints
func HTTPErrorHandler(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		// Add trace ID to context
		traceID := r.Header.Get("X-Trace-ID")
		if traceID == "" {
			traceID = uuid.New().String()
		}
		ctx := context.WithValue(r.Context(), "traceID", traceID)
		r = r.WithContext(ctx)

		// Create response wrapper to intercept errors
		wrapped := &responseWriter{
			ResponseWriter: w,
			request:        r,
			traceID:        traceID,
		}

		// Handle panics
		defer func() {
			if err := recover(); err != nil {
				handlePanic(wrapped, err)
			}
		}()

		// Process request
		next.ServeHTTP(wrapped, r)
	})
}

// responseWriter wraps http.ResponseWriter to intercept errors
type responseWriter struct {
	http.ResponseWriter
	request    *http.Request
	traceID    string
	statusCode int
	written    bool
}

func (w *responseWriter) WriteHeader(code int) {
	if !w.written {
		w.statusCode = code
		w.ResponseWriter.WriteHeader(code)
		w.written = true
	}
}

func (w *responseWriter) Write(b []byte) (int, error) {
	if !w.written {
		w.WriteHeader(http.StatusOK)
	}
	return w.ResponseWriter.Write(b)
}

// handlePanic converts panics to proper error responses
func handlePanic(w *responseWriter, recovered interface{}) {
	// Log stack trace
	debug.PrintStack()

	appErr := apperrors.NewInternal("An unexpected error occurred. Please try again later.", w.traceID, nil)
	writeErrorResponse(w, appErr)
}

// CustomHTTPError handles gRPC gateway error responses
func CustomHTTPError(ctx context.Context, mux *runtime.ServeMux,
	marshaler runtime.Marshaler, w http.ResponseWriter, r *http.Request, err error) {

	// Extract trace ID
	traceID := r.Header.Get("X-Trace-ID")
	if traceID == "" {
		if span := trace.SpanFromContext(ctx); span.SpanContext().IsValid() {
			traceID = span.SpanContext().TraceID().String()
		} else {
			traceID = uuid.New().String()
		}
	}

	// Convert gRPC error to HTTP response
	st, _ := status.FromError(err)

	// Check if we have our custom error detail in status details
	for _, detail := range st.Details() {
		if errorDetail, ok := detail.(*errorspb.ErrorDetail); ok {
			// Update the error detail with current request context
			errorDetail.TraceId = traceID
			errorDetail.Instance = r.URL.Path

			// Convert to JSON and write response
			w.Header().Set("Content-Type", "application/problem+json")
			w.WriteHeader(runtime.HTTPStatusFromCode(st.Code()))

			// Create a simplified JSON response that matches RFC 7807
			response := map[string]interface{}{
				"type":      getTypeForCode(errorDetail.Code),
				"title":     errorDetail.Title,
				"status":    runtime.HTTPStatusFromCode(st.Code()),
				"detail":    errorDetail.Detail,
				"instance":  errorDetail.Instance,
				"traceId":   errorDetail.TraceId,
				"timestamp": errorDetail.Timestamp,
			}

			// Add field violations if present
			if len(errorDetail.FieldViolations) > 0 {
				violations := make([]map[string]interface{}, len(errorDetail.FieldViolations))
				for i, fv := range errorDetail.FieldViolations {
					violations[i] = map[string]interface{}{
						"field":   fv.Field,
						"code":    fv.Code,
						"message": fv.Description,
					}
				}
				response["errors"] = violations
			}

			// Add extensions if present
			if len(errorDetail.Extensions) > 0 {
				extensions := make(map[string]interface{})
				for k, v := range errorDetail.Extensions {
					// Convert Any to JSON
					if jsonBytes, err := protojson.Marshal(v); err == nil {
						var jsonData interface{}
						if err := json.Unmarshal(jsonBytes, &jsonData); err == nil {
							extensions[k] = jsonData
						}
					}
				}
				if len(extensions) > 0 {
					response["extensions"] = extensions
				}
			}

			if err := json.NewEncoder(w).Encode(response); err != nil {
				http.Error(w, `{"error": "Failed to encode error response"}`, 500)
			}
			return
		}
	}

	// Fallback: create new error response
	fallbackErr := apperrors.NewInternal(st.Message(), traceID, nil)
	fallbackErr.GRPCCode = st.Code()
	writeAppErrorResponse(w, fallbackErr, r.URL.Path)
}

// Helper functions
func getTypeForCode(code string) string {
	switch code {
	case errorspb.AppErrorCode_VALIDATION_FAILED.String():
		return "https://api.example.com/errors/validation-failed"
	case errorspb.AppErrorCode_RESOURCE_NOT_FOUND.String():
		return "https://api.example.com/errors/resource-not-found"
	case errorspb.AppErrorCode_RESOURCE_CONFLICT.String():
		return "https://api.example.com/errors/resource-conflict"
	case errorspb.AppErrorCode_PERMISSION_DENIED.String():
		return "https://api.example.com/errors/permission-denied"
	case errorspb.AppErrorCode_INTERNAL_ERROR.String():
		return "https://api.example.com/errors/internal-error"
	case errorspb.AppErrorCode_SERVICE_UNAVAILABLE.String():
		return "https://api.example.com/errors/service-unavailable"
	default:
		return "https://api.example.com/errors/unknown"
	}
}

func writeErrorResponse(w http.ResponseWriter, err error) {
	if appErr, ok := err.(*apperrors.AppError); ok {
		writeAppErrorResponse(w, appErr, "")
	} else {
		http.Error(w, err.Error(), http.StatusInternalServerError)
	}
}

func writeAppErrorResponse(w http.ResponseWriter, appErr *apperrors.AppError, instance string) {
	statusCode := runtime.HTTPStatusFromCode(appErr.GRPCCode)

	response := map[string]interface{}{
		"type":      getTypeForCode(appErr.AppCode.String()),
		"title":     appErr.Title,
		"status":    statusCode,
		"detail":    appErr.Detail,
		"traceId":   appErr.TraceID,
		"timestamp": time.Now(),
	}

	if instance != "" {
		response["instance"] = instance
	}

	if len(appErr.FieldViolations) > 0 {
		violations := make([]map[string]interface{}, len(appErr.FieldViolations))
		for i, fv := range appErr.FieldViolations {
			violations[i] = map[string]interface{}{
				"field":   fv.Field,
				"code":    fv.Code,
				"message": fv.Description,
			}
		}
		response["errors"] = violations
	}

	w.Header().Set("Content-Type", "application/problem+json")
	w.WriteHeader(statusCode)
	json.NewEncoder(w).Encode(response)
}

package service

import (
	"context"
	"fmt"
	todopb "github.com/bhatti/todo-api-errors/api/proto/todo/v1"
	"github.com/bhatti/todo-api-errors/internal/errors"
	"github.com/bhatti/todo-api-errors/internal/repository"
	"github.com/bhatti/todo-api-errors/internal/validation"
	"github.com/google/uuid"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/trace"
	"google.golang.org/protobuf/types/known/fieldmaskpb"
	"google.golang.org/protobuf/types/known/timestamppb"
	"strings"
)

var tracer = otel.Tracer("todo-service")

// TodoService implements the TODO API
type TodoService struct {
	todopb.UnimplementedTodoServiceServer
	repo repository.TodoRepository
}

// NewTodoService creates a new TODO service
func NewTodoService(repo repository.TodoRepository) (*TodoService, error) {
	return &TodoService{
		repo: repo,
	}, nil
}

// CreateTask creates a new task
func (s *TodoService) CreateTask(ctx context.Context, req *todopb.CreateTaskRequest) (*todopb.Task, error) {
	ctx, span := tracer.Start(ctx, "CreateTask")
	defer span.End()

	// Get trace ID for error responses
	traceID := span.SpanContext().TraceID().String()

	// Validate request
	if req.Task == nil {
		return nil, errors.NewRequiredField("task", "Task object is required", traceID)
	}

	// Validate task fields using the new validation package
	if err := validation.ValidateTask(req.Task, traceID); err != nil {
		span.SetAttributes(attribute.String("validation.error", err.Error()))
		return nil, err
	}

	// Check for duplicate title
	existing, err := s.repo.GetTaskByTitle(ctx, req.Task.Title)
	if err != nil && !repository.IsNotFound(err) {
		span.RecordError(err)
		return nil, s.handleRepositoryError(err, traceID)
	}

	if existing != nil {
		return nil, errors.NewConflict("task", "A task with this title already exists", traceID)
	}

	// Generate task ID
	taskID := uuid.New().String()
	task := &todopb.Task{
		Name:        fmt.Sprintf("tasks/%s", taskID),
		Title:       req.Task.Title,
		Description: req.Task.Description,
		Status:      req.Task.Status,
		Priority:    req.Task.Priority,
		DueDate:     req.Task.DueDate,
		Tags:        req.Task.Tags,
		CreateTime:  timestamppb.Now(),
		UpdateTime:  timestamppb.Now(),
		CreatedBy:   s.getUserFromContext(ctx),
	}

	// Set defaults
	if task.Status == todopb.Status_STATUS_UNSPECIFIED {
		task.Status = todopb.Status_STATUS_PENDING
	}
	if task.Priority == todopb.Priority_PRIORITY_UNSPECIFIED {
		task.Priority = todopb.Priority_PRIORITY_MEDIUM
	}

	// Save to repository
	if err := s.repo.CreateTask(ctx, task); err != nil {
		span.RecordError(err)
		return nil, s.handleRepositoryError(err, traceID)
	}

	span.SetAttributes(
		attribute.String("task.id", taskID),
		attribute.String("task.title", task.Title),
	)

	return task, nil
}

// GetTask retrieves a specific task
func (s *TodoService) GetTask(ctx context.Context, req *todopb.GetTaskRequest) (*todopb.Task, error) {
	ctx, span := tracer.Start(ctx, "GetTask")
	defer span.End()

	traceID := span.SpanContext().TraceID().String()

	// Validate request using the new validation package
	if err := validation.ValidateRequest(req, traceID); err != nil {
		return nil, err
	}

	// Extract task ID
	parts := strings.Split(req.Name, "/")
	if len(parts) != 2 || parts[0] != "tasks" {
		return nil, errors.NewRequiredField("name", "Task name must be in format 'tasks/{id}'", traceID)
	}

	taskID := parts[1]
	span.SetAttributes(attribute.String("task.id", taskID))

	// Get from repository
	task, err := s.repo.GetTask(ctx, taskID)
	if err != nil {
		if repository.IsNotFound(err) {
			return nil, errors.NewNotFound("Task", taskID, traceID)
		}
		span.RecordError(err)
		return nil, s.handleRepositoryError(err, traceID)
	}

	// Check permissions
	if !s.canAccessTask(ctx, task) {
		return nil, errors.NewPermissionDenied("task", "read", traceID)
	}

	return task, nil
}

// ListTasks retrieves all tasks
func (s *TodoService) ListTasks(ctx context.Context, req *todopb.ListTasksRequest) (*todopb.ListTasksResponse, error) {
	ctx, span := tracer.Start(ctx, "ListTasks")
	defer span.End()

	traceID := span.SpanContext().TraceID().String()

	// Validate request using the new validation package
	if err := validation.ValidateRequest(req, traceID); err != nil {
		return nil, err
	}

	// Default page size
	pageSize := req.PageSize
	if pageSize == 0 {
		pageSize = 50
	}
	if pageSize > 1000 {
		pageSize = 1000
	}

	span.SetAttributes(
		attribute.Int("page.size", int(pageSize)),
		attribute.String("filter", req.Filter),
	)

	// Parse filter
	filter, err := s.parseFilter(req.Filter)
	if err != nil {
		return nil, errors.NewRequiredField("filter", fmt.Sprintf("Failed to parse filter: %v", err), traceID)
	}

	// Get tasks from repository
	tasks, nextPageToken, err := s.repo.ListTasks(ctx, repository.ListOptions{
		PageSize:  int(pageSize),
		PageToken: req.PageToken,
		Filter:    filter,
		OrderBy:   req.OrderBy,
		UserID:    s.getUserFromContext(ctx),
	})

	if err != nil {
		span.RecordError(err)
		return nil, s.handleRepositoryError(err, traceID)
	}

	// Get total count
	totalSize, err := s.repo.CountTasks(ctx, filter, s.getUserFromContext(ctx))
	if err != nil {
		// Log but don't fail the request
		span.RecordError(err)
		totalSize = -1
	}

	return &todopb.ListTasksResponse{
		Tasks:         tasks,
		NextPageToken: nextPageToken,
		TotalSize:     int32(totalSize),
	}, nil
}

// UpdateTask updates an existing task
func (s *TodoService) UpdateTask(ctx context.Context, req *todopb.UpdateTaskRequest) (*todopb.Task, error) {
	ctx, span := tracer.Start(ctx, "UpdateTask")
	defer span.End()

	traceID := span.SpanContext().TraceID().String()

	// Validate request
	if req.Task == nil {
		return nil, errors.NewRequiredField("task", "Task object is required", traceID)
	}

	if req.UpdateMask == nil || len(req.UpdateMask.Paths) == 0 {
		return nil, errors.NewRequiredField("update_mask", "Update mask must specify which fields to update", traceID)
	}

	// Extract task ID
	parts := strings.Split(req.Task.Name, "/")
	if len(parts) != 2 || parts[0] != "tasks" {
		return nil, errors.NewRequiredField("task.name", "Invalid task name format", traceID)
	}

	taskID := parts[1]
	span.SetAttributes(attribute.String("task.id", taskID))

	// Get existing task
	existing, err := s.repo.GetTask(ctx, taskID)
	if err != nil {
		if repository.IsNotFound(err) {
			return nil, errors.NewNotFound("Task", taskID, traceID)
		}
		return nil, s.handleRepositoryError(err, traceID)
	}

	// Check permissions
	if !s.canModifyTask(ctx, existing) {
		return nil, errors.NewPermissionDenied("task", "update", traceID)
	}

	// Apply updates based on field mask
	updated := s.applyFieldMask(existing, req.Task, req.UpdateMask)
	updated.UpdateTime = timestamppb.Now()

	// Validate updated task using the new validation package
	if err := validation.ValidateTask(updated, traceID); err != nil {
		return nil, err
	}

	// Save to repository
	if err := s.repo.UpdateTask(ctx, updated); err != nil {
		span.RecordError(err)
		return nil, s.handleRepositoryError(err, traceID)
	}

	return updated, nil
}

// DeleteTask removes a task
func (s *TodoService) DeleteTask(ctx context.Context, req *todopb.DeleteTaskRequest) (*todopb.DeleteTaskResponse, error) {
	ctx, span := tracer.Start(ctx, "DeleteTask")
	defer span.End()

	traceID := span.SpanContext().TraceID().String()

	// Validate request using the new validation package
	if err := validation.ValidateRequest(req, traceID); err != nil {
		return nil, err
	}

	// Extract task ID
	parts := strings.Split(req.Name, "/")
	if len(parts) != 2 || parts[0] != "tasks" {
		return nil, errors.NewRequiredField("name", "Invalid task name format", traceID)
	}

	taskID := parts[1]
	span.SetAttributes(attribute.String("task.id", taskID))

	// Get existing task to check permissions
	existing, err := s.repo.GetTask(ctx, taskID)
	if err != nil {
		if repository.IsNotFound(err) {
			return nil, errors.NewNotFound("Task", taskID, traceID)
		}
		return nil, s.handleRepositoryError(err, traceID)
	}

	// Check permissions
	if !s.canModifyTask(ctx, existing) {
		return nil, errors.NewPermissionDenied("task", "delete", traceID)
	}

	// Delete from repository
	if err := s.repo.DeleteTask(ctx, taskID); err != nil {
		span.RecordError(err)
		return nil, s.handleRepositoryError(err, traceID)
	}

	return &todopb.DeleteTaskResponse{
		Message: fmt.Sprintf("Task %s deleted successfully", req.Name),
	}, nil
}

// BatchCreateTasks creates multiple tasks at once
func (s *TodoService) BatchCreateTasks(ctx context.Context, req *todopb.BatchCreateTasksRequest) (*todopb.BatchCreateTasksResponse, error) {
	ctx, span := tracer.Start(ctx, "BatchCreateTasks")
	defer span.End()

	traceID := span.SpanContext().TraceID().String()

	// Validate batch request using the new validation package
	if err := validation.ValidateBatchCreateTasks(req, traceID); err != nil {
		span.SetAttributes(attribute.String("validation.error", err.Error()))
		return nil, err
	}

	// Process each task
	var created []*todopb.Task
	var batchErrors []string

	for i, createReq := range req.Requests {
		task, err := s.CreateTask(ctx, createReq)
		if err != nil {
			// Collect errors for batch response
			batchErrors = append(batchErrors, fmt.Sprintf("Task %d: %s", i, err.Error()))
			continue
		}
		created = append(created, task)
	}

	// If all tasks failed, return error
	if len(created) == 0 && len(batchErrors) > 0 {
		return nil, errors.NewInternal("All batch operations failed", traceID, nil)
	}

	// Return partial success
	response := &todopb.BatchCreateTasksResponse{
		Tasks: created,
	}

	// Add partial errors to response metadata if any
	if len(batchErrors) > 0 {
		span.SetAttributes(
			attribute.Int("batch.total", len(req.Requests)),
			attribute.Int("batch.success", len(created)),
			attribute.Int("batch.failed", len(batchErrors)),
		)
	}

	return response, nil
}

// Helper methods

func (s *TodoService) handleRepositoryError(err error, traceID string) error {
	if repository.IsConnectionError(err) {
		return errors.NewServiceUnavailable("Unable to connect to the database. Please try again later.", traceID)
	}

	// Log internal error details
	span := trace.SpanFromContext(context.Background())
	if span != nil {
		span.RecordError(err)
	}

	return errors.NewInternal("An unexpected error occurred while processing your request", traceID, err)
}

func (s *TodoService) getUserFromContext(ctx context.Context) string {
	// In a real implementation, this would extract user info from auth context
	if user, ok := ctx.Value("user").(string); ok {
		return user
	}
	return "anonymous"
}

func (s *TodoService) canAccessTask(ctx context.Context, task *todopb.Task) bool {
	// In a real implementation, check if user can access this task
	user := s.getUserFromContext(ctx)
	return user == task.CreatedBy || user == "admin"
}

func (s *TodoService) canModifyTask(ctx context.Context, task *todopb.Task) bool {
	// In a real implementation, check if user can modify this task
	user := s.getUserFromContext(ctx)
	return user == task.CreatedBy || user == "admin"
}

func (s *TodoService) parseFilter(filter string) (map[string]interface{}, error) {
	// Simple filter parser - in production, use a proper parser
	parsed := make(map[string]interface{})

	if filter == "" {
		return parsed, nil
	}

	// Example: "status=COMPLETED AND priority=HIGH"
	parts := strings.Split(filter, " AND ")
	for _, part := range parts {
		kv := strings.Split(strings.TrimSpace(part), "=")
		if len(kv) != 2 {
			return nil, fmt.Errorf("invalid filter expression: %s", part)
		}

		key := strings.TrimSpace(kv[0])
		value := strings.Trim(strings.TrimSpace(kv[1]), "'\"")

		// Validate filter keys
		switch key {
		case "status", "priority", "created_by":
			parsed[key] = value
		default:
			return nil, fmt.Errorf("unknown filter field: %s", key)
		}
	}

	return parsed, nil
}

func (s *TodoService) applyFieldMask(existing, update *todopb.Task, mask *fieldmaskpb.FieldMask) *todopb.Task {
	result := *existing

	for _, path := range mask.Paths {
		switch path {
		case "title":
			result.Title = update.Title
		case "description":
			result.Description = update.Description
		case "status":
			result.Status = update.Status
		case "priority":
			result.Priority = update.Priority
		case "due_date":
			result.DueDate = update.DueDate
		case "tags":
			result.Tags = update.Tags
		}
	}
	return &result
}

package main

import (
	"context"
	"fmt"
	"log"
	"net"
	"net/http"
	"os"
	"os/signal"
	"syscall"
	"time"

	todopb "github.com/bhatti/todo-api-errors/api/proto/todo/v1"
	"github.com/bhatti/todo-api-errors/internal/middleware"
	"github.com/bhatti/todo-api-errors/internal/monitoring"
	"github.com/bhatti/todo-api-errors/internal/repository"
	"github.com/bhatti/todo-api-errors/internal/service"

	"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
	"github.com/prometheus/client_golang/prometheus/promhttp"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/grpc/reflection"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/encoding/protojson"
)

func main() {
	// Initialize monitoring
	if err := monitoring.InitOpenTelemetryMetrics(); err != nil {
		log.Printf("Failed to initialize OpenTelemetry metrics: %v", err)
		// Continue without OpenTelemetry - Prometheus will still work
	}

	// Initialize repository
	repo := repository.NewInMemoryRepository()

	// Initialize service
	todoService, err := service.NewTodoService(repo)
	if err != nil {
		log.Fatalf("Failed to create service: %v", err)
	}

	// Start gRPC server
	grpcPort := ":50051"
	go func() {
		if err := startGRPCServer(grpcPort, todoService); err != nil {
			log.Fatalf("Failed to start gRPC server: %v", err)
		}
	}()

	// Start HTTP gateway
	httpPort := ":8080"
	go func() {
		if err := startHTTPGateway(httpPort, grpcPort); err != nil {
			log.Fatalf("Failed to start HTTP gateway: %v", err)
		}
	}()

	// Start metrics server
	go func() {
		http.Handle("/metrics", promhttp.Handler())
		if err := http.ListenAndServe(":9090", nil); err != nil {
			log.Printf("Failed to start metrics server: %v", err)
		}
	}()

	log.Printf("TODO API server started")
	log.Printf("gRPC server listening on %s", grpcPort)
	log.Printf("HTTP gateway listening on %s", httpPort)
	log.Printf("Metrics available at :9090/metrics")

	// Wait for interrupt signal
	sigCh := make(chan os.Signal, 1)
	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
	<-sigCh

	log.Println("Shutting down...")
}

func startGRPCServer(port string, todoService todopb.TodoServiceServer) error {
	lis, err := net.Listen("tcp", port)
	if err != nil {
		return fmt.Errorf("failed to listen: %w", err)
	}

	// Create gRPC server with interceptors - now using the new UnaryErrorInterceptor
	opts := []grpc.ServerOption{
		grpc.ChainUnaryInterceptor(
			middleware.UnaryErrorInterceptor, // Using new protobuf-based error interceptor
			loggingInterceptor(),
			recoveryInterceptor(),
		),
	}

	server := grpc.NewServer(opts...)

	// Register service
	todopb.RegisterTodoServiceServer(server, todoService)

	// Register reflection for debugging
	reflection.Register(server)

	return server.Serve(lis)
}

func startHTTPGateway(httpPort, grpcPort string) error {
	ctx := context.Background()

	// Create gRPC connection
	conn, err := grpc.DialContext(
		ctx,
		"localhost"+grpcPort,
		grpc.WithTransportCredentials(insecure.NewCredentials()),
	)
	if err != nil {
		return fmt.Errorf("failed to dial gRPC server: %w", err)
	}

	// Create gateway mux with custom error handler
	mux := runtime.NewServeMux(
		runtime.WithErrorHandler(middleware.CustomHTTPError), // Using new protobuf-based error handler
		runtime.WithMarshalerOption(runtime.MIMEWildcard, &runtime.JSONPb{
			MarshalOptions: protojson.MarshalOptions{
				UseProtoNames:   true,
				EmitUnpopulated: false,
			},
			UnmarshalOptions: protojson.UnmarshalOptions{
				DiscardUnknown: true,
			},
		}),
	)

	// Register service handler
	if err := todopb.RegisterTodoServiceHandler(ctx, mux, conn); err != nil {
		return fmt.Errorf("failed to register service handler: %w", err)
	}

	// Create HTTP server with middleware
	handler := middleware.HTTPErrorHandler( // Using new protobuf-based HTTP error handler
		corsMiddleware(
			authMiddleware(
				loggingHTTPMiddleware(mux),
			),
		),
	)

	server := &http.Server{
		Addr:         httpPort,
		Handler:      handler,
		ReadTimeout:  10 * time.Second,
		WriteTimeout: 10 * time.Second,
		IdleTimeout:  120 * time.Second,
	}

	return server.ListenAndServe()
}

// Middleware implementations

func loggingInterceptor() grpc.UnaryServerInterceptor {
	return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
		start := time.Now()

		// Call handler
		resp, err := handler(ctx, req)

		// Log request
		duration := time.Since(start)
		statusCode := "OK"
		if err != nil {
			statusCode = status.Code(err).String()
		}

		log.Printf("gRPC: %s %s %s %v", info.FullMethod, statusCode, duration, err)

		return resp, err
	}
}

func recoveryInterceptor() grpc.UnaryServerInterceptor {
	return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
		defer func() {
			if r := recover(); r != nil {
				log.Printf("Recovered from panic: %v", r)
				monitoring.RecordPanicRecovery(ctx)
				err = status.Error(codes.Internal, "Internal server error")
			}
		}()

		return handler(ctx, req)
	}
}

func loggingHTTPMiddleware(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		start := time.Now()

		// Wrap response writer to capture status
		wrapped := &statusResponseWriter{ResponseWriter: w, statusCode: http.StatusOK}

		// Process request
		next.ServeHTTP(wrapped, r)

		// Log request
		duration := time.Since(start)
		log.Printf("HTTP: %s %s %d %v", r.Method, r.URL.Path, wrapped.statusCode, duration)
	})
}

func corsMiddleware(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Access-Control-Allow-Origin", "*")
		w.Header().Set("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS, PATCH")
		w.Header().Set("Access-Control-Allow-Headers", "Content-Type, Authorization, X-Trace-ID")

		if r.Method == "OPTIONS" {
			w.WriteHeader(http.StatusOK)
			return
		}

		next.ServeHTTP(w, r)
	})
}

func authMiddleware(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		// Simple auth for demo - in production use proper authentication
		authHeader := r.Header.Get("Authorization")
		if authHeader == "" {
			authHeader = "Bearer anonymous"
		}

		// Extract user from token
		user := "anonymous"
		if len(authHeader) > 7 && authHeader[:7] == "Bearer " {
			user = authHeader[7:]
		}

		// Add user to context
		ctx := context.WithValue(r.Context(), "user", user)
		next.ServeHTTP(w, r.WithContext(ctx))
	})
}

type statusResponseWriter struct {
	http.ResponseWriter
	statusCode int
}

func (w *statusResponseWriter) WriteHeader(code int) {
	w.statusCode = code
	w.ResponseWriter.WriteHeader(code)
}

curl -X POST http://localhost:8080/v1/tasks \
-H "Content-Type: application/json" \
-d '{
"task": {
"title": "",
"description": "This description is wayyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyy too long…",
"status": "INVALID_STATUS",
"tags": ["INVALID TAG", "tag-1", "tag-1"]
}
}'

< HTTP/1.1 422 Unprocessable Entity
< Content-Type: application/problem+json
{
  "detail": "The request contains 5 validation errors",
  "errors": [
    {
      "code": "TOO_SHORT",
      "field": "title",
      "message": "value length must be at least 1 characters"
    },
    {
      "code": "TOO_LONG",
      "field": "description",
      "message": "value length must be at most 100 characters"
    },
    {
      "code": "INVALID_FORMAT",
      "field": "tags",
      "message": "value does not match regex pattern `^[a-z0-9-]+$`"
    },
    {
      "code": "INVALID_TAG_FORMAT",
      "field": "tags[0]",
      "message": "Tag 'INVALID TAG' must be lowercase letters, numbers, and hyphens only"
    },
    {
      "code": "DUPLICATE_TAG",
      "field": "tags[2]",
      "message": "Tag 'tag-1' appears multiple times"
    }
  ],
  "instance": "/v1/tasks",
  "status": 400,
  "timestamp": {
    "seconds": 1755288524,
    "nanos": 484865000
  },
  "title": "Validation Failed",
  "traceId": "eb4bfb3f-9397-4547-8618-ce9952a16067",
  "type": "https://api.example.com/errors/validation-failed"
}

curl http://localhost:8080/v1/tasks/non-existent-id

< HTTP/1.1 404 Not Found
< Content-Type: application/problem+json
{
  "detail": "Task with ID 'non-existent-id' was not found.",
  "instance": "/v1/tasks/non-existent-id",
  "status": 404,
  "timestamp": {
    "seconds": 1755288565,
    "nanos": 904607000
  },
  "title": "Resource Not Found",
  "traceId": "6ce00cd8-d0b7-47f1-b6f6-9fc1375c26a4",
  "type": "https://api.example.com/errors/resource-not-found"
}

curl -X POST http://localhost:8080/v1/tasks \
-H "Content-Type: application/json" \
-d '{
"task": {
"title": "Existing Task Title"
}
}'

curl -X POST http://localhost:8080/v1/tasks \
-H "Content-Type: application/json" \
-d '{
"task": {
"title": "Existing Task Title"
}
}'

< HTTP/1.1 409 Conflict
< Content-Type: application/problem+json
{
  "detail": "Conflict creating task: A task with this title already exists",
  "instance": "/v1/tasks",
  "status": 409,
  "timestamp": {
    "seconds": 1755288593,
    "nanos": 594458000
  },
  "title": "Resource Conflict",
  "traceId": "ed2e78d2-591d-492a-8d71-6b6843ce86f7",
  "type": "https://api.example.com/errors/resource-conflict"
}

curl http://localhost:8080/v1/tasks

HTTP/1.1 503 Service Unavailable
Content-Type: application/problem+json
Retry-After: 30
{
  "type": "https://api.example.com/errors/service-unavailable",
  "title": "Service Unavailable",
  "status": 503,
  "detail": "Database connection pool exhausted. Please try again later.",
  "instance": "/v1/tasks",
  "traceId": "db-pool-001",
  "timestamp": "2025-08-15T10:30:00Z",
  "extensions": {
    "retryable": true,
    "retryAfter": "2025-08-15T10:30:30Z",
    "maxRetries": 3,
    "backoffType": "exponential",
    "backoffMs": 1000,
    "errorCategory": "database"
  }
}

// DANGEROUS: Non-atomic operation
func (s *PaymentService) ProcessPayment(req *PaymentRequest) error {
    // Step 1: Debit source account
    if err := s.debitAccount(req.FromAccount, req.Amount); err != nil {
        return err
    }
    
    // ???? SIGKILL here leaves money debited but not credited
    // Step 2: Credit destination account  
    if err := s.creditAccount(req.ToAccount, req.Amount); err != nil {
        // Money is lost! Source debited but destination not credited
        return err
    }
    
    // Step 3: Record transaction
    return s.recordTransaction(req)
}

// DANGEROUS: Race condition during shutdown
func (s *InventoryService) ReserveItem(req *ReserveRequest) error {
    // Check availability
    if s.getStock(req.ItemID) < req.Quantity {
        return ErrInsufficientStock
    }
    
    // ???? SIGKILL here can cause double-reservation
    // Another request might see the same stock level
    
    // Reserve the item
    return s.updateStock(req.ItemID, -req.Quantity)
}

// Anti-entropy pattern for data consistency
type ReconciliationService struct {
    paymentDB    PaymentDatabase
    accountDB    AccountDatabase
    auditLog     AuditLogger
    alerting     AlertingService
}

func (r *ReconciliationService) ReconcilePayments(ctx context.Context) error {
    // Find payments without matching account entries
    orphanedPayments, err := r.paymentDB.FindOrphanedPayments(ctx)
    if err != nil {
        return err
    }
    
    for _, payment := range orphanedPayments {
        // Check if this was a partial transaction
        sourceDebit, _ := r.accountDB.GetTransaction(payment.FromAccount, payment.ID)
        destCredit, _ := r.accountDB.GetTransaction(payment.ToAccount, payment.ID)
        
        switch {
        case sourceDebit != nil && destCredit == nil:
            // Complete the transaction
            if err := r.creditAccount(payment.ToAccount, payment.Amount); err != nil {
                r.alerting.SendAlert("Failed to complete orphaned payment", payment.ID)
                continue
            }
            r.auditLog.RecordReconciliation("completed_payment", payment.ID)
            
        case sourceDebit == nil && destCredit != nil:
            // Reverse the credit
            if err := r.debitAccount(payment.ToAccount, payment.Amount); err != nil {
                r.alerting.SendAlert("Failed to reverse orphaned credit", payment.ID)
                continue
            }
            r.auditLog.RecordReconciliation("reversed_credit", payment.ID)
            
        default:
            // Both or neither exist - needs investigation
            r.alerting.SendAlert("Ambiguous payment state", payment.ID)
        }
    }
    
    return nil
}

// Run reconciliation periodically
func (r *ReconciliationService) Start(ctx context.Context) {
    ticker := time.NewTicker(5 * time.Minute)
    defer ticker.Stop()
    
    for {
        select {
        case <-ctx.Done():
            return
        case <-ticker.C:
            if err := r.ReconcilePayments(ctx); err != nil {
                log.Printf("Reconciliation failed: %v", err)
            }
        }
    }
}

//go:generate protoc --go_out=. --go_opt=paths=source_relative --go-grpc_out=. --go-grpc_opt=paths=source_relative api/demo.proto

package main

import (
    "context"
    "flag"
    "fmt"
    "log"
    "net"
    "net/http"
    "os"
    "os/signal"
    "sync/atomic"
    "syscall"
    "time"

    "google.golang.org/grpc"
    "google.golang.org/grpc/codes"
    health "google.golang.org/grpc/health/grpc_health_v1"
    "google.golang.org/grpc/status"
)

// Service represents our application with health state
type Service struct {
    isHealthy         atomic.Bool
    isShuttingDown    atomic.Bool
    activeRequests    atomic.Int64
    dependencyHealthy atomic.Bool
}

// HealthChecker implements the gRPC health checking protocol
type HealthChecker struct {
    svc *Service
}

func (h *HealthChecker) Check(ctx context.Context, req *health.HealthCheckRequest) (*health.HealthCheckResponse, error) {
    service := req.GetService()
    
    // Liveness: Simple check - is the process responsive?
    if service == "" || service == "liveness" {
        if h.svc.isShuttingDown.Load() {
            return &health.HealthCheckResponse{
                Status: health.HealthCheckResponse_NOT_SERVING,
            }, nil
        }
        return &health.HealthCheckResponse{
            Status: health.HealthCheckResponse_SERVING,
        }, nil
    }
    
    // Readiness: Deep check - can we handle traffic?
    if service == "readiness" {
        // Check application health
        if !h.svc.isHealthy.Load() {
            return &health.HealthCheckResponse{
                Status: health.HealthCheckResponse_NOT_SERVING,
            }, nil
        }
        
        // Check critical dependencies
        if !h.svc.dependencyHealthy.Load() {
            return &health.HealthCheckResponse{
                Status: health.HealthCheckResponse_NOT_SERVING,
            }, nil
        }
        
        // Check if shutting down
        if h.svc.isShuttingDown.Load() {
            return &health.HealthCheckResponse{
                Status: health.HealthCheckResponse_NOT_SERVING,
            }, nil
        }
        
        return &health.HealthCheckResponse{
            Status: health.HealthCheckResponse_SERVING,
        }, nil
    }
    
    // Synthetic readiness: Complex business logic check for monitoring
    if service == "synthetic-readiness" {
        // Simulate a complex health check that validates business logic
        // This would make actual API calls, database queries, etc.
        if !h.performSyntheticCheck(ctx) {
            return &health.HealthCheckResponse{
                Status: health.HealthCheckResponse_NOT_SERVING,
            }, nil
        }
        return &health.HealthCheckResponse{
            Status: health.HealthCheckResponse_SERVING,
        }, nil
    }
    
    return nil, status.Errorf(codes.NotFound, "unknown service: %s", service)
}

func (h *HealthChecker) performSyntheticCheck(ctx context.Context) bool {
    // In a real service, this would:
    // 1. Create a test transaction
    // 2. Query the database
    // 3. Call dependent services
    // 4. Validate the complete flow works
    return h.svc.isHealthy.Load() && h.svc.dependencyHealthy.Load()
}

func (h *HealthChecker) Watch(req *health.HealthCheckRequest, server health.Health_WatchServer) error {
    return status.Error(codes.Unimplemented, "watch not implemented")
}

// DemoServiceServer implements your business logic
type DemoServiceServer struct {
    UnimplementedDemoServiceServer
    svc *Service
}

func (s *DemoServiceServer) ProcessRequest(ctx context.Context, req *ProcessRequest) (*ProcessResponse, error) {
    s.svc.activeRequests.Add(1)
    defer s.svc.activeRequests.Add(-1)
    
    // Simulate processing
    select {
    case <-ctx.Done():
        return nil, ctx.Err()
    case <-time.After(100 * time.Millisecond):
        return &ProcessResponse{
            Result: fmt.Sprintf("Processed: %s", req.GetData()),
        }, nil
    }
}

func main() {
    var (
        port         = flag.Int("port", 8080, "gRPC port")
        mgmtPort     = flag.Int("mgmt-port", 8090, "Management port")
        startupDelay = flag.Duration("startup-delay", 10*time.Second, "Startup delay")
    )
    flag.Parse()
    
    svc := &Service{}
    svc.dependencyHealthy.Store(true) // Assume healthy initially
    
    // Management endpoints for testing
    mux := http.NewServeMux()
    mux.HandleFunc("/toggle-health", func(w http.ResponseWriter, r *http.Request) {
        current := svc.dependencyHealthy.Load()
        svc.dependencyHealthy.Store(!current)
        fmt.Fprintf(w, "Dependency health toggled to: %v\n", !current)
    })
    mux.HandleFunc("/metrics", func(w http.ResponseWriter, r *http.Request) {
        fmt.Fprintf(w, "active_requests %d\n", svc.activeRequests.Load())
        fmt.Fprintf(w, "is_healthy %v\n", svc.isHealthy.Load())
        fmt.Fprintf(w, "is_shutting_down %v\n", svc.isShuttingDown.Load())
    })
    
    mgmtServer := &http.Server{
        Addr:    fmt.Sprintf(":%d", *mgmtPort),
        Handler: mux,
    }
    
    // Start management server
    go func() {
        log.Printf("Management server listening on :%d", *mgmtPort)
        if err := mgmtServer.ListenAndServe(); err != http.ErrServerClosed {
            log.Fatalf("Management server failed: %v", err)
        }
    }()
    
    // Simulate slow startup
    log.Printf("Starting application (startup delay: %v)...", *startupDelay)
    time.Sleep(*startupDelay)
    svc.isHealthy.Store(true)
    log.Println("Application initialized and ready")
    
    // Setup gRPC server
    lis, err := net.Listen("tcp", fmt.Sprintf(":%d", *port))
    if err != nil {
        log.Fatalf("Failed to listen: %v", err)
    }
    
    grpcServer := grpc.NewServer()
    RegisterDemoServiceServer(grpcServer, &DemoServiceServer{svc: svc})
    health.RegisterHealthServer(grpcServer, &HealthChecker{svc: svc})
    
    // Start gRPC server
    go func() {
        log.Printf("gRPC server listening on :%d", *port)
        if err := grpcServer.Serve(lis); err != nil {
            log.Fatalf("gRPC server failed: %v", err)
        }
    }()
    
    // Wait for shutdown signal
    sigCh := make(chan os.Signal, 1)
    signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
    sig := <-sigCh
    
    log.Printf("Received signal: %v, starting graceful shutdown...", sig)
    
    // Graceful shutdown sequence
    svc.isShuttingDown.Store(true)
    svc.isHealthy.Store(false) // Fail readiness immediately
    
    // Stop accepting new requests
    grpcServer.GracefulStop()
    
    // Wait for active requests to complete
    timeout := time.After(30 * time.Second)
    ticker := time.NewTicker(100 * time.Millisecond)
    defer ticker.Stop()
    
    for {
        select {
        case <-timeout:
            log.Println("Shutdown timeout reached, forcing exit")
            os.Exit(1)
        case <-ticker.C:
            active := svc.activeRequests.Load()
            if active == 0 {
                log.Println("All requests completed")
                goto shutdown
            }
            log.Printf("Waiting for %d active requests to complete...", active)
        }
    }
    
shutdown:
    // Cleanup
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()
    mgmtServer.Shutdown(ctx)
    
    log.Println("Graceful shutdown complete")
}

apiVersion: apps/v1
kind: Deployment
metadata:
  name: bad-service
  namespace: demo
spec:
  replicas: 2
  selector:
    matchLabels:
      app: bad-service
  template:
    metadata:
      labels:
        app: bad-service
      # MISSING: Critical Istio annotations!
    spec:
      # DEFAULT: Only 30s grace period
      containers:
      - name: app
        image: myregistry/demo-service:latest
        ports:
        - containerPort: 8080
          name: grpc
        - containerPort: 8090
          name: mgmt
        args: ["--startup-delay=45s"]  # Longer than default probe timeout!
        
        # ANTI-PATTERN: Identical liveness and readiness probes
        livenessProbe:
          exec:
            command: ["/bin/grpc_health_probe", "-addr=:8080"]
          initialDelaySeconds: 10
          periodSeconds: 10
          failureThreshold: 3  # Will fail after 40s total
          
        readinessProbe:
          exec:
            command: ["/bin/grpc_health_probe", "-addr=:8080"]  # Same as liveness!
          initialDelaySeconds: 10
          periodSeconds: 10
        
        # MISSING: No startup probe for slow initialization
        # MISSING: No preStop hook for graceful shutdown

apiVersion: apps/v1
kind: Deployment
metadata:
  name: good-service
  namespace: demo
spec:
  replicas: 2
  selector:
    matchLabels:
      app: good-service
  template:
    metadata:
      labels:
        app: good-service
      annotations:
        # Critical for Istio/Envoy sidecar lifecycle management
        sidecar.istio.io/holdApplicationUntilProxyStarts: "true"
        proxy.istio.io/config: |
          proxyMetadata:
            EXIT_ON_ZERO_ACTIVE_CONNECTIONS: "true"
        sidecar.istio.io/proxyCPU: "100m"
        sidecar.istio.io/proxyMemory: "128Mi"
    spec:
      # Extended grace period: preStop (15s) + app shutdown (30s) + buffer (20s)
      terminationGracePeriodSeconds: 65
      
      containers:
      - name: app
        image: myregistry/demo-service:latest
        ports:
        - containerPort: 8080
          name: grpc
        - containerPort: 8090
          name: mgmt
        args: ["--startup-delay=45s"]
        
        # Resource management for predictable performance
        resources:
          requests:
            cpu: 100m
            memory: 128Mi
          limits:
            cpu: 500m
            memory: 512Mi
        
        # Startup probe for slow initialization
        startupProbe:
          exec:
            command: ["/bin/grpc_health_probe", "-addr=:8080", "-service=readiness"]
          initialDelaySeconds: 0
          periodSeconds: 5
          failureThreshold: 24  # 5s * 24 = 120s total startup time
          successThreshold: 1
        
        # Simple liveness check
        livenessProbe:
          exec:
            command: ["/bin/grpc_health_probe", "-addr=:8080", "-service=liveness"]
          initialDelaySeconds: 0  # Startup probe handles initialization
          periodSeconds: 10
          failureThreshold: 3
          timeoutSeconds: 5
        
        # Deep readiness check
        readinessProbe:
          exec:
            command: ["/bin/grpc_health_probe", "-addr=:8080", "-service=readiness"]
          initialDelaySeconds: 0
          periodSeconds: 5
          failureThreshold: 2
          successThreshold: 1
          timeoutSeconds: 5
        
        # Graceful shutdown coordination
        lifecycle:
          preStop:
            exec:
              command: ["/bin/sh", "-c", "sleep 15"]  # Allow LB to drain
        
        # Environment variables for cloud provider integration
        env:
        - name: CLOUD_PROVIDER
          value: "auto-detect"  # Works with GCP, AWS, Azure
        - name: ENABLE_PROFILING
          value: "true"

apiVersion: networking.istio.io/v1alpha3
kind: DestinationRule
metadata:
  name: payment-service
  namespace: demo
spec:
  host: payment-service.demo.svc.cluster.local
  trafficPolicy:
    connectionPool:
      tcp:
        maxConnections: 100
      http:
        http1MaxPendingRequests: 100
        maxRequestsPerConnection: 2
    circuitBreaker:
      consecutiveErrors: 5
      interval: 30s
      baseEjectionTime: 30s
      maxEjectionPercent: 50
    retryPolicy:
      attempts: 3
      perTryTimeout: 2s
      retryOn: 5xx,gateway-error,connect-failure,refused-stream
      retryRemoteLocalities: true

terminationGracePeriodSeconds = preStop delay + application shutdown timeout + buffer

Examples:
- Simple service: 10s + 20s + 5s = 35s
- Complex service: 15s + 45s + 5s = 65s
- Batch processor: 30s + 120s + 10s = 160s

package idempotency

import (
    "context"
    "crypto/sha256"
    "encoding/hex"
    "time"
    "sync"
    "errors"
)

var (
    ErrDuplicateRequest = errors.New("duplicate request detected")
    ErrProcessingInProgress = errors.New("request is currently being processed")
)

// IdempotencyStore tracks request execution with persistence
type IdempotencyStore struct {
    mu        sync.RWMutex
    records   map[string]*Record
    persister PersistenceLayer // Database or Redis for durability
}

type Record struct {
    Key         string
    Response    interface{}
    Error       error
    Status      ProcessingStatus
    ExpiresAt   time.Time
    CreatedAt   time.Time
    ProcessedAt *time.Time
}

type ProcessingStatus int

const (
    StatusPending ProcessingStatus = iota
    StatusProcessing
    StatusCompleted
    StatusFailed
)

// ProcessIdempotent ensures exactly-once processing semantics
func (s *IdempotencyStore) ProcessIdempotent(
    ctx context.Context,
    key string,
    ttl time.Duration,
    fn func() (interface{}, error),
) (interface{}, error) {
    // Check if we've seen this request before
    s.mu.RLock()
    record, exists := s.records[key]
    s.mu.RUnlock()
    
    if exists {
        switch record.Status {
        case StatusCompleted:
            if time.Now().Before(record.ExpiresAt) {
                return record.Response, record.Error
            }
        case StatusProcessing:
            return nil, ErrProcessingInProgress
        case StatusFailed:
            if time.Now().Before(record.ExpiresAt) {
                return record.Response, record.Error
            }
        }
    }
    
    // Mark as processing
    record = &Record{
        Key:       key,
        Status:    StatusProcessing,
        ExpiresAt: time.Now().Add(ttl),
        CreatedAt: time.Now(),
    }
    
    s.mu.Lock()
    s.records[key] = record
    s.mu.Unlock()
    
    // Persist the processing state
    if err := s.persister.Save(ctx, record); err != nil {
        return nil, err
    }
    
    // Execute the function
    response, err := fn()
    processedAt := time.Now()
    
    // Update record with result
    s.mu.Lock()
    record.Response = response
    record.Error = err
    record.ProcessedAt = &processedAt
    if err != nil {
        record.Status = StatusFailed
    } else {
        record.Status = StatusCompleted
    }
    s.mu.Unlock()
    
    // Persist the final state
    s.persister.Save(ctx, record)
    
    return response, err
}

// Example: Idempotent payment processing
func (s *PaymentService) ProcessPayment(ctx context.Context, req *PaymentRequest) (*PaymentResponse, error) {
    // Generate idempotency key from request
    key := generateIdempotencyKey(req)
    
    result, err := s.idempotencyStore.ProcessIdempotent(
        ctx,
        key,
        24*time.Hour, // Keep records for 24 hours
        func() (interface{}, error) {
            // Atomic transaction processing
            return s.processPaymentTransaction(ctx, req)
        },
    )
    
    if err != nil {
        return nil, err
    }
    return result.(*PaymentResponse), nil
}

// Atomic transaction processing
func (s *PaymentService) processPaymentTransaction(ctx context.Context, req *PaymentRequest) (*PaymentResponse, error) {
    // Use database transaction for atomicity
    tx, err := s.db.BeginTx(ctx, nil)
    if err != nil {
        return nil, err
    }
    defer tx.Rollback()
    
    // Step 1: Validate accounts
    if err := s.validateAccounts(ctx, tx, req); err != nil {
        return nil, err
    }
    
    // Step 2: Process payment atomically
    paymentID, err := s.executePayment(ctx, tx, req)
    if err != nil {
        return nil, err
    }
    
    // Step 3: Commit transaction
    if err := tx.Commit(); err != nil {
        return nil, err
    }
    
    return &PaymentResponse{
        PaymentID: paymentID,
        Status:    "completed",
        Timestamp: time.Now(),
    }, nil
}

package checkpoint

import (
    "context"
    "encoding/json"
    "time"
)

type CheckpointStore interface {
    Save(ctx context.Context, id string, state interface{}) error
    Load(ctx context.Context, id string, state interface{}) error
    Delete(ctx context.Context, id string) error
}

type BatchProcessor struct {
    store          CheckpointStore
    checkpointFreq int
}

type BatchState struct {
    JobID      string    `json:"job_id"`
    TotalItems int       `json:"total_items"`
    Processed  int       `json:"processed"`
    LastItem   string    `json:"last_item"`
    StartedAt  time.Time `json:"started_at"`
}

func (p *BatchProcessor) ProcessBatch(ctx context.Context, jobID string, items []string) error {
    // Try to resume from checkpoint
    state := &BatchState{JobID: jobID}
    if err := p.store.Load(ctx, jobID, state); err == nil {
        log.Printf("Resuming job %s from item %d", jobID, state.Processed)
        items = items[state.Processed:]
    } else {
        // New job
        state = &BatchState{
            JobID:      jobID,
            TotalItems: len(items),
            Processed:  0,
            StartedAt:  time.Now(),
        }
    }
    
    // Process items with periodic checkpointing
    for i, item := range items {
        select {
        case <-ctx.Done():
            // Save progress before shutting down
            state.LastItem = item
            return p.store.Save(ctx, jobID, state)
        default:
            // Process item
            if err := p.processItem(ctx, item); err != nil {
                return err
            }
            
            state.Processed++
            state.LastItem = item
            
            // Checkpoint periodically
            if state.Processed%p.checkpointFreq == 0 {
                if err := p.store.Save(ctx, jobID, state); err != nil {
                    log.Printf("Failed to checkpoint: %v", err)
                }
            }
        }
    }
    
    // Job completed, remove checkpoint
    return p.store.Delete(ctx, jobID)
}

package circuitbreaker

import (
    "context"
    "sync"
    "time"
)

type State int

const (
    StateClosed State = iota
    StateOpen
    StateHalfOpen
)

type CircuitBreaker struct {
    mu              sync.RWMutex
    state           State
    failures        int
    successes       int
    lastFailureTime time.Time
    
    maxFailures      int
    resetTimeout     time.Duration
    halfOpenRequests int
}

func (cb *CircuitBreaker) Call(ctx context.Context, fn func() error) error {
    cb.mu.RLock()
    state := cb.state
    cb.mu.RUnlock()
    
    if state == StateOpen {
        // Check if we should transition to half-open
        cb.mu.Lock()
        if time.Since(cb.lastFailureTime) > cb.resetTimeout {
            cb.state = StateHalfOpen
            cb.successes = 0
            state = StateHalfOpen
        }
        cb.mu.Unlock()
    }
    
    if state == StateOpen {
        return ErrCircuitOpen
    }
    
    err := fn()
    
    cb.mu.Lock()
    defer cb.mu.Unlock()
    
    if err != nil {
        cb.failures++
        cb.lastFailureTime = time.Now()
        
        if cb.failures >= cb.maxFailures {
            cb.state = StateOpen
            log.Printf("Circuit breaker opened after %d failures", cb.failures)
        }
        return err
    }
    
    if state == StateHalfOpen {
        cb.successes++
        if cb.successes >= cb.halfOpenRequests {
            cb.state = StateClosed
            cb.failures = 0
            log.Println("Circuit breaker closed")
        }
    }
    
    return nil
}

# Deploy both services
kubectl apply -f k8s/bad-service.yaml
kubectl apply -f k8s/good-service.yaml

# Watch pods in separate terminal
watch kubectl get pods -n demo

# Force restart
kubectl delete pod -l app=bad-service -n demo

# Observe: Pod will enter CrashLoopBackOff due to liveness probe
# killing it before 45s startup completes

# Force restart
kubectl delete pod -l app=good-service -n demo

# Observe: Pod stays in 0/1 Ready state for ~45s, then becomes ready
# No restarts occur thanks to startup probe

# Deploy payment service with reconciliation enabled
kubectl apply -f k8s/payment-service.yaml

# Start payment traffic generator
kubectl run payment-generator --image=payment-client:latest \
  --restart=Never --rm -it -- \
  --target=payment-service.demo.svc.cluster.local:8080 \
  --rate=10 --duration=60s

# In another terminal, kill pods abruptly
while true; do
  kubectl delete pod -l app=payment-service -n demo --force --grace-period=0
  sleep 30
done

# Check for data inconsistencies
kubectl logs -l app=payment-service -n demo | grep "inconsistency"

# Monitor reconciliation metrics
kubectl port-forward svc/payment-service 8090:8090
curl http://localhost:8090/metrics | grep consistency

# Simulate regional failure
kubectl patch deployment payment-service -n demo \
  --patch '{"spec":{"replicas":0}}'

# Measure RTO - time to restore service
start_time=$(date +%s)
kubectl patch deployment payment-service -n demo \
  --patch '{"spec":{"replicas":3}}'

# Wait for all pods to be ready
kubectl wait --for=condition=ready pod -l app=payment-service -n demo --timeout=900s
end_time=$(date +%s)
rto=$((end_time - start_time))

echo "RTO: ${rto} seconds"
if [ $rto -le 900 ]; then
  echo "? RTO target met (15 minutes)"
else
  echo "? RTO target exceeded"
fi

# Deploy with fault injection
kubectl apply -f istio/fault-injection.yaml

# Generate requests with chaos header
for i in {1..100}; do
  grpcurl -H "x-chaos-test: true" -plaintext \
    payment-service.demo.svc.cluster.local:8080 \
    PaymentService/ProcessPayment \
    -d '{"amount": 100, "currency": "USD"}'
done

# Check Istio metrics for retry behavior
kubectl exec -n istio-system deployment/istiod -- \
  pilot-agent request GET stats/prometheus | grep retry

package monitoring

import (
    "time"
    "github.com/prometheus/client_golang/prometheus"
    "github.com/prometheus/client_golang/prometheus/promauto"
)

var (
    // RTO-related metrics
    ServiceStartupTime = promauto.NewHistogramVec(prometheus.HistogramOpts{
        Name: "service_startup_duration_seconds",
        Help: "Time from pod start to service ready",
        Buckets: []float64{1, 5, 10, 30, 60, 120, 300, 600}, // Up to 10 minutes
    }, []string{"service", "version"})
    
    ServiceRecoveryTime = promauto.NewHistogramVec(prometheus.HistogramOpts{
        Name: "service_recovery_duration_seconds", 
        Help: "Time to recover from failure state",
        Buckets: []float64{1, 5, 10, 30, 60, 300, 900}, // Up to 15 minutes
    }, []string{"service", "failure_type"})
    
    // RPO-related metrics
    LastCheckpointAge = promauto.NewGaugeVec(prometheus.GaugeOpts{
        Name: "last_checkpoint_age_seconds",
        Help: "Age of last successful checkpoint",
    }, []string{"service", "checkpoint_type"})
    
    DataConsistencyChecks = promauto.NewCounterVec(prometheus.CounterOpts{
        Name: "data_consistency_checks_total",
        Help: "Total number of consistency checks performed",
    }, []string{"service", "check_type", "status"})
    
    InconsistencyDetected = promauto.NewCounterVec(prometheus.CounterOpts{
        Name: "data_inconsistencies_detected_total",
        Help: "Total number of data inconsistencies detected",
    }, []string{"service", "inconsistency_type", "severity"})
)

{
  "dashboard": {
    "title": "Service Lifecycle - Business Impact",
    "panels": [
      {
        "title": "RTO Compliance",
        "description": "Percentage of recoveries meeting RTO target (15 minutes)",
        "targets": [{
          "expr": "100 * (histogram_quantile(0.95, service_recovery_duration_seconds_bucket) <= 900)"
        }],
        "thresholds": [
          {"value": 95, "color": "green"},
          {"value": 90, "color": "yellow"},
          {"value": 0, "color": "red"}
        ]
      },
      {
        "title": "RPO Risk Assessment",
        "description": "Data at risk based on checkpoint age",
        "targets": [{
          "expr": "last_checkpoint_age_seconds / 60"
        }],
        "unit": "minutes"
      },
      {
        "title": "Data Consistency Status",
        "targets": [{
          "expr": "rate(data_inconsistencies_detected_total[5m])"
        }]
      }
    ]
  }
}

startupProbe:
  httpGet:
    path: /healthz
    port: 8080
  failureThreshold: 30  # 30 * 10s = 5 minutes
  periodSeconds: 10

// Implement atomic operations with database transactions
tx, err := db.BeginTx(ctx, nil)
if err != nil {
    return err
}
defer tx.Rollback()

// All operations within transaction
if err := processPayment(tx, req); err != nil {
    return err // Automatic rollback
}

return tx.Commit()

annotations:
  sidecar.istio.io/holdApplicationUntilProxyStarts: "true"
  proxy.istio.io/config: |
    proxyMetadata:
      EXIT_ON_ZERO_ACTIVE_CONNECTIONS: "true"