package main
import (
    "encoding/json"
    "os"
    "github.com/golang-migrate/migrate/v4"
    "github.com/golang-migrate/migrate/v4/database/postgres"
    _ "github.com/golang-migrate/migrate/v4/source/file"
    "database/sql"
    env "github.com/Netflix/go-env"
    _ "github.com/lib/pq"
    "log"
    "net/http"
    "github.com/gorilla/handlers"
    "github.com/gorilla/mux"
    "github.com/streadway/amqp"
)

type environment struct {
    PgsqlURI  string `env:"PGSQL_URI"`
    Listen    string `env:"LISTEN"`
    RabbitURI string `env:"RABBIT_URI"`
}

func main() {
var err error
// Getting configuration
log.Printf("INFO: Getting environment variables\n")
cnf := environment{}
_, err = env.UnmarshalFromEnviron(&cnf)
if err != nil {
    log.Fatal(err)
}
// Connecting to database
log.Printf("INFO: Connecting to database")
db, err = sql.Open("postgres", cnf.PgsqlURI)
if err != nil {
    log.Fatalf("Can't connect to postgresql: %v", err)
}
// Running migrations
driver, err := postgres.WithInstance(db, &postgres.Config{})
if err != nil {
    log.Fatalf("Can't get postgres driver: %v", err)
}
m, err := migrate.NewWithDatabaseInstance("file://./migrations", "postgres", driver)
if err != nil {
    log.Fatalf("Can't get migration object: %v", err)
}
m.Up()

// Initialising rabbit mq
// Initing rabbitmq
conn, err := amqp.Dial(cnf.RabbitURI)
if err != nil {
    log.Fatalf("Can't connect to rabbitmq")
}
defer conn.Close()
ch, err = conn.Channel()
if err != nil {
    log.Fatalf("Can't open channel")
}
defer ch.Close()
err = initRabbit()
if err != nil {
    log.Fatalf("Can't create rabbitmq queues: %s\n", err)
}

// Setting handlers for query
log.Printf("INFO: Starting listening on %s\n", cnf.Listen)
router := mux.NewRouter().StrictSlash(true)
// PROJECTS
router.HandleFunc("/requests", authMiddleware(getRequests)).Methods("GET")
router.HandleFunc("/requests", authMiddleware(addRequest)).Methods("POST")
router.HandleFunc("/requests/{name}", authMiddleware(getRequest)).Methods("GET")
router.HandleFunc("/requests/{name}", authMiddleware(updRequest)).Methods("PUT")
router.HandleFunc("/requests/{name}", authMiddleware(delRequest)).Methods("DELETE")
http.ListenAndServe(cnf.Listen, handlers.LoggingHandler(os.Stdout, router))

func authMiddleware(next http.HandlerFunc) http.HandlerFunc {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        tokenString := r.Header.Get("X-API-KEY")
        if tokenString != "804b95f13b714ee9912b19861faf3d25" {
            w.WriteHeader(http.StatusUnauthorized)
            w.Write([]byte("Missing Authorization Header\n"))
            return
        }
        next(w, r)
    })
}

func initRabbit() error {
    err := ch.ExchangeDeclare(
        "VideoParserExchange", // name
        "fanout",              // type
        true,                  // durable
        false,                 // auto delete
        false,                 // internal
        false,                 // no wait
        nil,                   // arguments
    )
    if err != nil {
        return err
    }
    err = ch.ExchangeDeclare(
        "VideoParserRetryExchange", // name
        "fanout",                   // type
        true,                       // durable
        false,                      // auto delete
        false,                      // internal
        false,                      // no wait
        nil,                        // arguments
    )
    if err != nil {
        return err
    }

args := amqp.Table{"x-dead-letter-exchange": "VideoParserRetryExchange"}
    queue, err = ch.QueueDeclare(
        "VideoParserWorkerQueue", // name
        true,                     // durable - flush to disk
        false,                    // delete when unused
        false,                    // exclusive - only accessible by the connection that declares
        false,                    // no-wait - the queue will assume to be declared on the server
        args,                     // arguments -
    )
    if err != nil {
        return err
    }
    args = amqp.Table{"x-dead-letter-exchange": "VideoParserExchange", "x-message-ttl": 60000}
    queue, err = ch.QueueDeclare(
        "VideoParserWorkerRetryQueue", // name
        true,                          // durable - flush to disk
        false,                         // delete when unused
        false,                         // exclusive - only accessible by the connection that declares
        false,                         // no-wait - the queue will assume to be declared on the server
        args,                          // arguments -
    )
    if err != nil {
        return err
    }
    queue, err = ch.QueueDeclare(
        "VideoParserArchiveQueue", // name
        true,                      // durable - flush to disk
        false,                     // delete when unused
        false,                     // exclusive - only accessible by the connection that declares
        false,                     // no-wait - the queue will assume to be declared on the server
        nil,                       // arguments -
    )
    if err != nil {
        return err
    }

err = ch.QueueBind("VideoParserWorkerQueue", "*", "VideoParserExchange", false, nil)
    if err != nil {
        return err
    }
    err = ch.QueueBind("VideoParserArchiveQueue", "*", "VideoParserExchange", false, nil)
    if err != nil {
        return err
    }
    err = ch.QueueBind("VideoParserWorkerRetryQueue", "*", "VideoParserRetryExchange", false, nil)
    if err != nil {
        return err
    }
    return nil
}

CREATE TABLE IF NOT EXISTS requests (
    id SERIAL,
    name VARCHAR(256),
    description VARCHAR(2048),
    video_url VARCHAR(64),
    text_url VARCHAR(64),
    processed BOOL DEFAULT FALSE,
    archived BOOL DEFAULT FALSE,
    created_at TIMESTAMP DEFAULT now(),
    updated_at TIMESTAMP DEFAULT null,
    UNIQUE(name)
);

DROP TABLE requests;

func addRequest(w http.ResponseWriter, r *http.Request) {
    // Parsing event
    req := postRequestRequest{}
    err := json.NewDecoder(r.Body).Decode(&req)
    if err != nil {
        log.Printf("WARNING: Can't parse incoming request: %s\n", err)
        returnResponse(400, "Can't parse json", nil, w)
        return
    }
    request := Request{}
    if req.Name == nil {
        returnResponse(400, "name can't be null", nil, w)
        return
    }
    request.Name = *req.Name
    if req.Description != nil {
        request.Description = *req.Description
    }
    if req.Processed != nil {
        request.Processed = *req.Processed
    }
    if req.VideoURL != nil {
        request.VideoURL = *req.VideoURL
    }
    if req.TextURL != nil {
        request.TextURL = *req.TextURL
    }
    // Publishing data to rabbitmq
    msg, err := json.Marshal(request)
    if err != nil {
        log.Printf("ERROR: Marshaling request: %s\n", err)
        returnResponse(500, "Can't marshal request ", nil, w)
        return
    }
    err = ch.Publish(
        "VideoParserExchange", // exchange
        "",                    // routing key
        false,                 // mandatory - could return an error if there are no consumers or queue
        false,                 // immediate
        amqp.Publishing{
            DeliveryMode: amqp.Persistent,
            ContentType:  "application/json",
            Body:         msg,
        })
    if err != nil {
        log.Printf("ERROR: Publishing to rabbit: %s\n", err)
        returnResponse(500, "Can't publish to rabbit ", nil, w)
        return
    }
    stmt := `INSERT INTO requests (name, description, processed, video_url, text_url) VALUES ($1, $2, $3, $4, $5) RETURNING id`
    err = db.QueryRow(stmt, &request.Name, &request.Description, &request.Processed, &request.VideoURL, &request.TextURL).Scan(&request.ID)
    if err != nil {
        log.Printf("ERROR: Adding new request to database: %s\n", err)
        returnResponse(500, "Can't add new request ", nil, w)
        return
    }
    returnResponse(200, "Successfully added new request", nil, w)
}

FROM golang:1.15-alpine AS build
# Installing requirements
RUN apk add --update git && \
    rm -rf /tmp/* /var/tmp/* /var/cache/apk/* /var/cache/distfiles/*
# Creating workdir and copying dependencies
WORKDIR /go/src/app
COPY . .
# Installing dependencies
RUN go get
ENV CGO_ENABLED=0
RUN go build -o api main.go requests.go
FROM alpine:3.9.6
RUN echo "http://dl-cdn.alpinelinux.org/alpine/edge/testing/" >> /etc/apk/repositories && \
    apk add --update bash && \
    rm -rf /tmp/* /var/tmp/* /var/cache/apk/* /var/cache/distfiles/*
WORKDIR /app
COPY --from=build /go/src/app/api /app/api
COPY ./migrations/ /app/migrations/
CMD ["/app/api"]

mkdir helm
cd helm
helm repo add bitnami https://charts.bitnami.com/bitnami
helm search repo bitnami

helm pull bitnami/rabbitmq
tar zxv

cp rabbitmq/values.yaml ./values.dev.yaml
vi values.dev.yaml

kubectl create ns stage
helm instal -n stage rabbitmq -f values.dev.yaml
mv values.dev.yaml values. stage. yaml
helm install -n stage rabbitmq -f values.stage.yanl ./rabbitmq/

kubectl -n stage apply -f deployment.yaml
kubectl -n stage get deploy

curl -H 'X-API-KEY: 804b95f13b714ee9912b19861faf3d25' -s http://api.stage.kis.im/requests | jq .

curl -X POST -d '{"name": "test1", "video_url": "https://google.com" }' -H 'X-API-KEY: 804b95f13b714ee9912b19861faf3d25' -s http://api.stage.kis.im/requests | jq .

curl -H 'X-API-KEY: 804b95f13b714ee9912b19861faf3d25' -s
http://api.stage.kis.im/requests/request1 | jq .