1package main
  2
  3import (
  4	"context"
  5	"database/sql"
  6	"fmt"
  7	"log/slog"
  8	"os"
  9
 10	"github.com/jmoiron/sqlx"
 11	"github.com/picosh/pico/pkg/db"
 12	"github.com/picosh/pico/pkg/db/postgres"
 13)
 14
 15func findPosts(dbpool *sqlx.DB) ([]*db.Post, error) {
 16	var posts []*db.Post
 17	rs, err := dbpool.Query(`SELECT
 18		posts.id, user_id, filename, title, text, description,
 19		posts.created_at, publish_at, posts.updated_at, hidden, COALESCE(views, 0) as views
 20		FROM posts
 21		LEFT OUTER JOIN post_analytics ON post_analytics.post_id = posts.id
 22	`)
 23	if err != nil {
 24		return posts, err
 25	}
 26	for rs.Next() {
 27		post := &db.Post{}
 28		err := rs.Scan(
 29			&post.ID,
 30			&post.UserID,
 31			&post.Filename,
 32			&post.Title,
 33			&post.Text,
 34			&post.Description,
 35			&post.CreatedAt,
 36			&post.PublishAt,
 37			&post.UpdatedAt,
 38			&post.Hidden,
 39			&post.Views,
 40		)
 41		if err != nil {
 42			return posts, err
 43		}
 44
 45		posts = append(posts, post)
 46	}
 47	if rs.Err() != nil {
 48		return posts, rs.Err()
 49	}
 50	return posts, nil
 51}
 52
 53func insertUser(tx *sql.Tx, user *db.User) error {
 54	_, err := tx.Exec(
 55		"INSERT INTO app_users (id, name, created_at) VALUES($1, $2, $3)",
 56		user.ID,
 57		user.Name,
 58		user.CreatedAt,
 59	)
 60	return err
 61}
 62
 63func insertPublicKey(tx *sql.Tx, pk *db.PublicKey) error {
 64	_, err := tx.Exec(
 65		"INSERT INTO public_keys (id, user_id, public_key, created_at) VALUES ($1, $2, $3, $4)",
 66		pk.ID,
 67		pk.UserID,
 68		pk.Key,
 69		pk.CreatedAt,
 70	)
 71	return err
 72}
 73
 74func insertPost(tx *sql.Tx, post *db.Post) error {
 75	_, err := tx.Exec(
 76		`INSERT INTO posts
 77			(id, user_id, title, text, created_at, publish_at, updated_at, description, filename, hidden, cur_space, views)
 78			VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)`,
 79		post.ID,
 80		post.UserID,
 81		post.Title,
 82		post.Text,
 83		post.CreatedAt,
 84		post.PublishAt,
 85		post.UpdatedAt,
 86		post.Description,
 87		post.Filename,
 88		post.Hidden,
 89		post.Space,
 90		post.Views,
 91	)
 92	return err
 93}
 94
 95type ConflictData struct {
 96	User          *db.User
 97	Pks           []*db.PublicKey
 98	ReplaceWithID string
 99}
100
101func main() {
102	logger := slog.Default()
103
104	listsDb := postgres.NewDB(os.Getenv("LISTS_DB_URL"), logger)
105
106	proseDb := postgres.NewDB(os.Getenv("PROSE_DB_URL"), logger)
107
108	picoDb := postgres.NewDB(os.Getenv("PICO_DB_URL"), logger)
109
110	ctx := context.Background()
111	tx, err := picoDb.Db.BeginTx(ctx, nil)
112	if err != nil {
113		panic(err)
114	}
115	defer func() {
116		err = tx.Rollback()
117	}()
118
119	logger.Info("Finding prose users")
120	proseUsers, err := proseDb.FindUsers()
121	if err != nil {
122		panic(err)
123	}
124
125	logger.Info("Finding lists users")
126	listUsers, err := listsDb.FindUsers()
127	if err != nil {
128		panic(err)
129	}
130
131	logger.Info("Adding prose users and public keys to PICO db")
132	userMap := map[string]*db.User{}
133	for _, proseUser := range proseUsers {
134		userMap[proseUser.Name] = proseUser
135
136		err = insertUser(tx, proseUser)
137		if err != nil {
138			panic(err)
139		}
140
141		proseKeys, err := proseDb.FindKeysByUser(proseUser)
142		if err != nil {
143			panic(err)
144		}
145
146		for _, prosePK := range proseKeys {
147			err = insertPublicKey(tx, prosePK)
148			if err != nil {
149				panic(err)
150			}
151		}
152	}
153
154	noconflicts := []*ConflictData{}
155	conflicts := []*ConflictData{}
156	updateIDs := []*ConflictData{}
157	logger.Info("Finding conflicts")
158	for _, listUser := range listUsers {
159		listKeys, err := listsDb.FindKeysByUser(listUser)
160		if err != nil {
161			panic(err)
162		}
163
164		data := &ConflictData{
165			User: listUser,
166			Pks:  listKeys,
167		}
168
169		if userMap[listUser.Name] == nil {
170			noconflicts = append(noconflicts, data)
171			continue
172		} else {
173			proseUser := userMap[listUser.Name]
174			proseKeys, err := proseDb.FindKeysByUser(proseUser)
175			if err != nil {
176				panic(err)
177			}
178
179			if len(listKeys) != len(proseKeys) {
180				conflicts = append(conflicts, data)
181				continue
182			}
183
184			pkMap := map[string]bool{}
185			for _, prosePK := range proseKeys {
186				pkMap[prosePK.Key] = true
187			}
188
189			conflicted := false
190			for _, listPK := range listKeys {
191				if !pkMap[listPK.Key] {
192					conflicted = true
193					conflicts = append(conflicts, data)
194					break
195				}
196			}
197
198			if !conflicted {
199				data.ReplaceWithID = proseUser.ID
200				updateIDs = append(updateIDs, data)
201			}
202		}
203	}
204
205	logger.Info("adding records with no conflicts", "len", len(noconflicts))
206	for _, data := range noconflicts {
207		err = insertUser(tx, data.User)
208		if err != nil {
209			panic(err)
210		}
211
212		for _, pk := range data.Pks {
213			err = insertPublicKey(tx, pk)
214			if err != nil {
215				panic(err)
216			}
217		}
218	}
219
220	logger.Info("adding records with conflicts", "len", len(conflicts))
221	for _, data := range conflicts {
222		data.User.Name = fmt.Sprintf("%stmp", data.User.Name)
223		err = insertUser(tx, data.User)
224		if err != nil {
225			panic(err)
226		}
227
228		for _, pk := range data.Pks {
229			err = insertPublicKey(tx, pk)
230			if err != nil {
231				panic(err)
232			}
233		}
234	}
235
236	prosePosts, err := findPosts(proseDb.Db)
237	if err != nil {
238		panic(err)
239	}
240
241	logger.Info("Adding posts from prose.sh")
242	for _, post := range prosePosts {
243		post.Space = "prose"
244		err = insertPost(tx, post)
245		if err != nil {
246			panic(err)
247		}
248	}
249
250	listPosts, err := findPosts(listsDb.Db)
251	if err != nil {
252		panic(err)
253	}
254
255	logger.Info("Adding posts from lists.sh")
256	for _, post := range listPosts {
257		updated := false
258		for _, alreadyAdded := range updateIDs {
259			if post.UserID == alreadyAdded.User.ID {
260				// we need to change the ID for these posts to the prose user id
261				// because we were able to determine it was the same user
262				post.UserID = alreadyAdded.ReplaceWithID
263				post.Space = "lists"
264				err = insertPost(tx, post)
265				if err != nil {
266					panic(err)
267				}
268				updated = true
269				break
270			}
271		}
272
273		if updated {
274			continue
275		}
276
277		post.Space = "lists"
278		err = insertPost(tx, post)
279		if err != nil {
280			panic(err)
281		}
282	}
283
284	logger.Info("Committing transactions to PICO db")
285	// Commit the transaction.
286	if err = tx.Commit(); err != nil {
287		panic(err)
288	}
289}