package main

import (
	"context"
	"errors"
	"log"
	"time"

	"github.com/jmoiron/sqlx"
        
	_ "github.com/vancelongwill/ksql-go/stdlib" // import the database/sql driver
)

type Item struct {
        K  string `db:"K"`
        V1 int    `db:"V1"`
        V2 string `db:"V2"`
        V3 bool   `db:"V3"`
}

func run(ctx context.Context) error {
	db, err := sqlx.Open("ksqldb", "http://0.0.0.0:8088/")
	if err != nil {
		return err
	}
	rows, err := db.QueryxContext(ctx, "SELECT * FROM t1 WHERE v1 > -1 EMIT CHANGES;")
	if err != nil {
		return err
	}
	defer rows.Close()
	// this will continue forever unless the context is cancelled, or rows.Close is called
	for rows.Next() {
                var r Item
		err := rows.StructScan(&r)
		if err != nil {
			return err
		}
		log.Println(r)
	}
	return rows.Err()
}

	// create a ksqldb client with a custom HTTP client (for auth etc)
	httpClient := &http.Client{}
	client := ksql.New("http://0.0.0.0:8088", ksql.WithHTTPClient(httpClient))

        // convert to databse/sql connection
	sqlDB := sql.OpenDB(stdlib.NewConnector(client))
	// again, using jmoiron/sqlx to interact with the DB
	db := sqlx.NewDb(sqlDB, "ksqldb")

type basicAuth struct {
	Username string
	Password string
}

func (bat basicAuth) RoundTrip(req *http.Request) (*http.Response, error) {
	req.Header.Set("Authorization", fmt.Sprintf("Basic %s",
		base64.StdEncoding.EncodeToString([]byte(fmt.Sprintf("%s:%s",
			bat.Username, bat.Password)))))
	return http.DefaultTransport.RoundTrip(req)
}

func run(ctx context.Context) error {
	// create a ksqldb client with a custom HTTP client (for auth etc)
	httpClient := &http.Client{
		Transport: basicAuth{
			Username: "[email protected]",
			Password: "somepassword",
		},
	}
	client := ksql.New("http://0.0.0.0:8088", ksql.WithHTTPClient(httpClient))

	sqlDB := sql.OpenDB(stdlib.NewConnector(client))
	db := sqlx.NewDb(sqlDB, "ksqldb")
        ...
}

	creds := &clientcredentials.Config{
		ClientID:     "yourID",
		ClientSecret: "yoursecret",
		TokenURL:     "https://yourauthservice.com/oauth2/token",
	}
	client := ksql.New("http://0.0.0.0:8088", ksql.WithHTTPClient(creds.Client(ctx)))