package sink

import (
	"database/sql"
	_ "github.com/lib/pq"
)

type PgSink struct {
	db *sql.DB
	in  chan interface{}
}


func NewPgSink(connStr string) (*PgSink,error) {
	var err error
	var db *sql.DB
	db, err = sql.Open("postgres", connStr)
	if err != nil {
		return nil, err
	}
	pg := &PgSink{
		db: db,
		in: make(chan interface{}),
	}
	go pg.init()
	return pg,nil
}

func (pg *PgSink) init()  {
	for  msg := range pg.in {
		switch m := msg.(type) {
		case string:
			_, _ = pg.db.Exec(m)
		}
	}
}

// In returns channel for receiving data
func (pg *PgSink) In() chan<- interface{} {
	return pg.in
}