agenda/storage/mongodb.go

133 lines
3.1 KiB
Go

package storage
import (
"context"
"fmt"
"github.com/spf13/viper"
"go.mongodb.org/mongo-driver/mongo"
"go.mongodb.org/mongo-driver/mongo/options"
"gopkg.in/mgo.v2/bson"
)
type MongoDBStorage struct {
*mongo.Client
DbName string
Collections map[string]string
}
func NewMongoDBStorage(cfg *viper.Viper) (MongoDBStorage, error) {
var (
mongodb_host = cfg.GetString("storage.db.mongodb.host")
mongodb_port = cfg.GetString("storage.db.mongodb.port")
mongodb_dbname = cfg.GetString("storage.db.mongodb.db_name")
mongodb_events = cfg.GetString("storage.db.mongodb.collections.events")
)
client, err := mongo.NewClient(options.Client().ApplyURI("mongodb://" + mongodb_host + ":" + mongodb_port))
if err != nil {
return MongoDBStorage{}, err
}
err = client.Connect(context.TODO())
if err != nil {
return MongoDBStorage{}, err
}
storage := MongoDBStorage{
Client: client,
DbName: mongodb_dbname,
Collections: map[string]string{
"events": mongodb_events,
},
}
//TODO Indexes
return storage, err
}
func (s MongoDBStorage) CreateEvent(event Event) error {
collection := s.Client.Database(s.DbName).Collection(s.Collections["events"])
if _, err := collection.InsertOne(context.TODO(), event); err != nil {
return err
}
return nil
}
func (s MongoDBStorage) GetEvent(id string) (*Event, error) {
collection := s.Client.Database(s.DbName).Collection(s.Collections["events"])
event := &Event{}
if err := collection.FindOne(context.TODO(), bson.M{"_id": id}).Decode(event); err != nil {
return nil, err
}
return event, nil
}
func (s MongoDBStorage) GetEvents(namespaces []string) (events []Event, err error) {
collection := s.Client.Database(s.DbName).Collection(s.Collections["events"])
var cur *mongo.Cursor
findOptions := options.Find()
//findOptions.SetSort(bson.D{{"startdate", 1}})
if len(namespaces) == 0 {
cur, err = collection.Find(context.TODO(), bson.D{}, findOptions)
if err != nil {
fmt.Println("...")
fmt.Println(err)
return events, err
}
} else {
cur, err = collection.Find(context.TODO(), bson.M{"namespace": bson.M{"$in": namespaces}}, findOptions)
if err != nil {
fmt.Println("Error here")
fmt.Println(err)
return events, err
}
}
for cur.Next(context.TODO()) {
var event Event
//var elem bson.M
if err := cur.Decode(&event); err != nil {
return events, err
}
// fmt.Println(elem)
// bsonBytes, _ := bson.Marshal(elem)
// fmt.Println(string(bsonBytes))
// bson.Unmarshal(bsonBytes, &event)
events = append(events, event)
}
return
}
func (s MongoDBStorage) AddSubscription(eventid string, subscription Subscription) error {
collection := s.Client.Database(s.DbName).Collection(s.Collections["events"])
filter := bson.M{"_id": eventid}
push := bson.M{"$push": bson.M{"subscriptions": subscription}}
_, err := collection.UpdateOne(context.TODO(), filter, push)
return err
}
func (s MongoDBStorage) UpdateEvent(event Event) error {
collection := s.Client.Database(s.DbName).Collection(s.Collections["events"])
if _, err := collection.ReplaceOne(context.TODO(), bson.M{"_id": event.ID}, event); err != nil {
fmt.Println(err)
return err
}
return nil
}